diff --git a/CHANGELOG.md b/CHANGELOG.md index d636190..65f8de0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,9 +13,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed - Except for `requests_auth.testing`, only direct access via `requests_auth.` was considered publicly exposed. This is now explicit, as inner packages are now using private prefix (`_`). If you were relying on some classes or functions that are now internal, feel free to open an issue. +- `requests_auth.JsonTokenFileCache` and `requests_auth.TokenMemoryCache` `get_token` method does not handle kwargs anymore, the `on_missing_token` callable does not expect any arguments anymore. +- `requests_auth.JsonTokenFileCache` does not expose `tokens_path` or `last_save_time` attributes anymore and is also allowing `pathlib.Path` instances as cache location. +- `requests_auth.TokenMemoryCache` does not expose `forbid_concurrent_cache_access` or `forbid_concurrent_missing_token_function_call` attributes anymore. ### Fixed -- Type information is now provided following [PEP 561](https://www.python.org/dev/peps/pep-0561/) +- Type information is now provided following [PEP 561](https://www.python.org/dev/peps/pep-0561/). +- Remove deprecation warnings due to usage of `utcnow` and `utcfromtimestamp`. +- Tokens cache `DEBUG` logs will not display tokens anymore. ### Removed - Removing support for Python `3.7`. diff --git a/README.md b/README.md index 72d02f5..15c3164 100644 --- a/README.md +++ b/README.md @@ -806,7 +806,7 @@ import datetime from requests_auth.testing import browser_mock, BrowserMock, create_token def test_something(browser_mock: BrowserMock): - token_expiry = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + token_expiry = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(hours=1) token = create_token(token_expiry) tab = browser_mock.add_response( opened_url="http://url_opened_by_browser?state=1234", diff --git a/requests_auth/_oauth2/implicit.py b/requests_auth/_oauth2/implicit.py index 25d588b..d971718 100644 --- a/requests_auth/_oauth2/implicit.py +++ b/requests_auth/_oauth2/implicit.py @@ -109,12 +109,14 @@ def __call__(self, r): token = OAuth2.token_cache.get_token( key=self.state, early_expiry=self.early_expiry, - on_missing_token=authentication_responses_server.request_new_grant, - grant_details=self.grant_details, + on_missing_token=self.request_new_token, ) r.headers[self.header_name] = self.header_value.format(token=token) return r + def request_new_token(self) -> tuple: + return authentication_responses_server.request_new_grant(self.grant_details) + class AzureActiveDirectoryImplicit(OAuth2Implicit): """ diff --git a/requests_auth/_oauth2/tokens.py b/requests_auth/_oauth2/tokens.py index a27826f..5d65609 100644 --- a/requests_auth/_oauth2/tokens.py +++ b/requests_auth/_oauth2/tokens.py @@ -4,6 +4,8 @@ import datetime import threading import logging +from pathlib import Path + from requests_auth._errors import * logger = logging.getLogger(__name__) @@ -23,16 +25,15 @@ def _decode_base64(base64_encoded_string: str) -> str: def _is_expired(expiry: float, early_expiry: float) -> bool: - return ( - datetime.datetime.utcfromtimestamp(expiry - early_expiry) - < datetime.datetime.utcnow() - ) + return datetime.datetime.fromtimestamp( + expiry - early_expiry, datetime.timezone.utc + ) < datetime.datetime.now(datetime.timezone.utc) def _to_expiry(expires_in: Union[int, str]) -> float: - expiry = datetime.datetime.utcnow().replace( - tzinfo=datetime.timezone.utc - ) + datetime.timedelta(seconds=int(expires_in)) + expiry = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta( + seconds=int(expires_in) + ) return expiry.timestamp() @@ -43,8 +44,8 @@ class TokenMemoryCache: def __init__(self): self.tokens = {} - self.forbid_concurrent_cache_access = threading.Lock() - self.forbid_concurrent_missing_token_function_call = threading.Lock() + self._forbid_concurrent_cache_access = threading.Lock() + self._forbid_concurrent_missing_token_function_call = threading.Lock() def _add_bearer_token(self, key: str, token: str): """ @@ -92,11 +93,11 @@ def _add_token( :param expiry: UTC timestamp of expiry :param refresh_token: refresh token value """ - with self.forbid_concurrent_cache_access: + with self._forbid_concurrent_cache_access: self.tokens[key] = token, expiry, refresh_token self._save_tokens() logger.debug( - f'Inserting token expiring on {datetime.datetime.utcfromtimestamp(expiry)} (UTC) with "{key}" key: {token}' + f'Inserting token expiring on {datetime.datetime.fromtimestamp(expiry, datetime.timezone.utc)} with "{key}" key.' ) def get_token( @@ -106,7 +107,6 @@ def get_token( early_expiry: float = 30.0, on_missing_token=None, on_expired_token=None, - **on_missing_token_kwargs, ) -> str: """ Return the bearer token. @@ -118,13 +118,12 @@ def get_token( expired 30 seconds before real expiry by default. :param on_missing_token: function to call when token is expired or missing (returning token and expiry tuple) :param on_expired_token: function to call to refresh the token when it is expired - :param on_missing_token_kwargs: arguments of the on_missing_token function (key-value arguments) :return: the token :raise AuthenticationFailed: in case token cannot be retrieved. """ logger.debug(f'Retrieving token with "{key}" key.') refresh_token = None - with self.forbid_concurrent_cache_access: + with self._forbid_concurrent_cache_access: self._load_tokens() if key in self.tokens: token = self.tokens[key] @@ -137,23 +136,23 @@ def get_token( del self.tokens[key] else: logger.debug( - f"Using already received authentication, will expire on {datetime.datetime.utcfromtimestamp(expiry)} (UTC)." + f"Using already received authentication, will expire on {datetime.datetime.fromtimestamp(expiry, datetime.timezone.utc)}." ) return bearer if refresh_token is not None and on_expired_token is not None: try: - with self.forbid_concurrent_missing_token_function_call: + with self._forbid_concurrent_missing_token_function_call: state, token, expires_in, refresh_token = on_expired_token( refresh_token ) self._add_access_token(state, token, expires_in, refresh_token) logger.debug(f"Refreshed token with key {key}.") - with self.forbid_concurrent_cache_access: + with self._forbid_concurrent_cache_access: if state in self.tokens: bearer, expiry, refresh_token = self.tokens[state] logger.debug( - f"Using newly refreshed token, expiring on {datetime.datetime.utcfromtimestamp(expiry)} (UTC)." + f"Using newly refreshed token, expiring on {datetime.datetime.fromtimestamp(expiry, datetime.timezone.utc)}." ) return bearer except (InvalidGrantRequest, GrantNotProvided): @@ -161,8 +160,8 @@ def get_token( logger.debug("Token cannot be found in cache.") if on_missing_token is not None: - with self.forbid_concurrent_missing_token_function_call: - new_token = on_missing_token(**on_missing_token_kwargs) + with self._forbid_concurrent_missing_token_function_call: + new_token = on_missing_token() if len(new_token) == 2: # Bearer token state, token = new_token self._add_bearer_token(state, token) @@ -176,21 +175,21 @@ def get_token( logger.warning( f"Using a token received on another key than expected. Expecting {key} but was {state}." ) - with self.forbid_concurrent_cache_access: + with self._forbid_concurrent_cache_access: if state in self.tokens: bearer, expiry, refresh_token = self.tokens[state] logger.debug( - f"Using newly received authentication, expiring on {datetime.datetime.utcfromtimestamp(expiry)} (UTC)." + f"Using newly received authentication, expiring on {datetime.datetime.fromtimestamp(expiry, datetime.timezone.utc)}." ) return bearer logger.debug( - f"User was not authenticated: key {key} cannot be found in {self.tokens}." + f"User was not authenticated: key {key} cannot be found in {list(self.tokens)}." ) raise AuthenticationFailed() def clear(self): - with self.forbid_concurrent_cache_access: + with self._forbid_concurrent_cache_access: logger.debug("Clearing token cache.") self.tokens = {} self._clear() @@ -210,36 +209,36 @@ class JsonTokenFileCache(TokenMemoryCache): Class to manage tokens using a cache file. """ - def __init__(self, tokens_path: str): + def __init__(self, tokens_path: Union[str, Path]): TokenMemoryCache.__init__(self) - self.tokens_path = tokens_path - self.last_save_time = 0 + self._tokens_path = Path(tokens_path) + self._last_save_time = 0 self._load_tokens() def _clear(self): - self.last_save_time = 0 + self._last_save_time = 0 try: - os.remove(self.tokens_path) + self._tokens_path.unlink(missing_ok=True) except: logger.debug("Cannot remove tokens file.") def _save_tokens(self): try: - with open(self.tokens_path, "w") as tokens_cache_file: + with self._tokens_path.open(mode="w") as tokens_cache_file: json.dump(self.tokens, tokens_cache_file) - self.last_save_time = os.path.getmtime(self.tokens_path) + self._last_save_time = os.path.getmtime(self._tokens_path) except: logger.exception("Cannot save tokens.") def _load_tokens(self): - if not os.path.exists(self.tokens_path): + if not self._tokens_path.exists(): logger.debug("No token loaded. Token cache does not exists.") return try: - last_modification_time = os.path.getmtime(self.tokens_path) - if last_modification_time > self.last_save_time: - self.last_save_time = last_modification_time - with open(self.tokens_path, "r") as tokens_cache_file: + last_modification_time = os.path.getmtime(self._tokens_path) + if last_modification_time > self._last_save_time: + self._last_save_time = last_modification_time + with self._tokens_path.open(mode="r") as tokens_cache_file: self.tokens = json.load(tokens_cache_file) except: logger.exception("Cannot load tokens.") diff --git a/tests/test_add_operator.py b/tests/test_add_operator.py index c83a589..6ff87f1 100644 --- a/tests/test_add_operator.py +++ b/tests/test_add_operator.py @@ -356,7 +356,9 @@ def test_oauth2_implicit_and_api_key_authentication_can_be_combined( token_cache, responses: RequestsMock, browser_mock: BrowserMock ): implicit_auth = requests_auth.OAuth2Implicit("http://provide_token") - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token = create_token(expiry_in_1_hour) tab = browser_mock.add_response( opened_url="http://provide_token?response_type=token&state=42a85b271b7a652ca3cc4c398cfd3f01b9ad36bf9c945ba823b023e8f8b95c4638576a0e3dcc96838b838bec33ec6c0ee2609d62ed82480b3b8114ca494c0521&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F", @@ -376,7 +378,9 @@ def test_oauth2_implicit_and_multiple_authentication_can_be_combined( token_cache, responses: RequestsMock, browser_mock: BrowserMock ): implicit_auth = requests_auth.OAuth2Implicit("http://provide_token") - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token = create_token(expiry_in_1_hour) tab = browser_mock.add_response( opened_url="http://provide_token?response_type=token&state=42a85b271b7a652ca3cc4c398cfd3f01b9ad36bf9c945ba823b023e8f8b95c4638576a0e3dcc96838b838bec33ec6c0ee2609d62ed82480b3b8114ca494c0521&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F", diff --git a/tests/test_and_operator.py b/tests/test_and_operator.py index 984cf18..f6f920a 100644 --- a/tests/test_and_operator.py +++ b/tests/test_and_operator.py @@ -356,7 +356,9 @@ def test_oauth2_implicit_and_api_key_authentication_can_be_combined( token_cache, responses: RequestsMock, browser_mock: BrowserMock ): implicit_auth = requests_auth.OAuth2Implicit("http://provide_token") - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token = create_token(expiry_in_1_hour) tab = browser_mock.add_response( opened_url="http://provide_token?response_type=token&state=42a85b271b7a652ca3cc4c398cfd3f01b9ad36bf9c945ba823b023e8f8b95c4638576a0e3dcc96838b838bec33ec6c0ee2609d62ed82480b3b8114ca494c0521&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F", @@ -376,7 +378,9 @@ def test_oauth2_implicit_and_multiple_authentication_can_be_combined( token_cache, responses: RequestsMock, browser_mock: BrowserMock ): implicit_auth = requests_auth.OAuth2Implicit("http://provide_token") - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token = create_token(expiry_in_1_hour) tab = browser_mock.add_response( opened_url="http://provide_token?response_type=token&state=42a85b271b7a652ca3cc4c398cfd3f01b9ad36bf9c945ba823b023e8f8b95c4638576a0e3dcc96838b838bec33ec6c0ee2609d62ed82480b3b8114ca494c0521&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F", diff --git a/tests/test_json_token_file_cache.py b/tests/test_json_token_file_cache.py index c722f02..6913cd7 100644 --- a/tests/test_json_token_file_cache.py +++ b/tests/test_json_token_file_cache.py @@ -1,4 +1,6 @@ import datetime +import logging +import pathlib import pytest import jwt @@ -8,18 +10,22 @@ @pytest.fixture -def token_cache(request) -> requests_auth.JsonTokenFileCache: - _token_cache = requests_auth.JsonTokenFileCache(request.node.name + ".cache") +def token_cache(tmp_path) -> requests_auth.JsonTokenFileCache: + _token_cache = requests_auth.JsonTokenFileCache(tmp_path / "my_tokens.cache") yield _token_cache _token_cache.clear() def test_add_bearer_tokens(token_cache): - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token1 = jwt.encode({"exp": expiry_in_1_hour}, "secret") token_cache._add_bearer_token("key1", token1) - expiry_in_2_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=2) + expiry_in_2_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=2) token2 = jwt.encode({"exp": expiry_in_2_hour}, "secret") token_cache._add_bearer_token("key2", token2) @@ -32,45 +38,90 @@ def test_add_bearer_tokens(token_cache): assert token_cache.get_token("key2") == token2 -def test_save_bearer_tokens(token_cache, request): - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) +def test_save_bearer_tokens(token_cache, tmp_path): + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token1 = jwt.encode({"exp": expiry_in_1_hour}, "secret") token_cache._add_bearer_token("key1", token1) - expiry_in_2_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=2) + expiry_in_2_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=2) token2 = jwt.encode({"exp": expiry_in_2_hour}, "secret") token_cache._add_bearer_token("key2", token2) - same_cache = requests_auth.JsonTokenFileCache(request.node.name + ".cache") + same_cache = requests_auth.JsonTokenFileCache(tmp_path / "my_tokens.cache") assert same_cache.get_token("key1") == token1 assert same_cache.get_token("key2") == token2 -def test_save_bearer_token_exception_handling(token_cache, request, monkeypatch): +def test_save_bearer_token_exception_handling( + token_cache, tmp_path, monkeypatch, caplog +): def failing_dump(*args): raise Exception("Failure") monkeypatch.setattr(requests_auth._oauth2.tokens.json, "dump", failing_dump) - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token1 = jwt.encode({"exp": expiry_in_1_hour}, "secret") + caplog.set_level(logging.DEBUG) + # Assert that the exception is not thrown token_cache._add_bearer_token("key1", token1) - same_cache = requests_auth.JsonTokenFileCache(request.node.name + ".cache") + same_cache = requests_auth.JsonTokenFileCache(tmp_path / "my_tokens.cache") with pytest.raises(requests_auth.AuthenticationFailed) as exception_info: same_cache.get_token("key1") assert str(exception_info.value) == "User was not authenticated." + assert caplog.messages == [ + "Cannot save tokens.", + f'Inserting token expiring on {expiry_in_1_hour:%Y-%m-%d %H:%M:%S+00:00} with "key1" key.', + "Cannot load tokens.", + 'Retrieving token with "key1" key.', + "Token cannot be found in cache.", + "User was not authenticated: key key1 cannot be found in [].", + ] + + +def test_missing_token_on_empty_cache(token_cache, caplog): + caplog.set_level(logging.DEBUG) + with pytest.raises(requests_auth.AuthenticationFailed): + token_cache.get_token("key1") + assert caplog.messages == [ + 'Retrieving token with "key1" key.', + "No token loaded. Token cache does not exists.", + "Token cannot be found in cache.", + "User was not authenticated: key key1 cannot be found in [].", + ] + + +def test_missing_token_on_non_empty_cache(token_cache, caplog): + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) + token1 = jwt.encode({"exp": expiry_in_1_hour}, "secret") + token_cache._add_bearer_token("key0", token1) -def test_missing_token(token_cache): + caplog.set_level(logging.DEBUG) with pytest.raises(requests_auth.AuthenticationFailed): token_cache.get_token("key1") + assert caplog.messages == [ + 'Retrieving token with "key1" key.', + "Token cannot be found in cache.", + "User was not authenticated: key key1 cannot be found in ['key0'].", + ] def test_missing_token_function(token_cache): - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token = jwt.encode({"exp": expiry_in_1_hour}, "secret") retrieved_token = token_cache.get_token( "key1", on_missing_token=lambda: ("key1", token) @@ -79,15 +130,30 @@ def test_missing_token_function(token_cache): def test_token_without_refresh_token(token_cache): - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) # add token without refresh token token = jwt.encode({"exp": expiry_in_1_hour}, "secret") token_cache.tokens["key1"] = ( token, - expiry_in_1_hour.replace(tzinfo=datetime.timezone.utc).timestamp(), + expiry_in_1_hour.timestamp(), ) token_cache._save_tokens() # try to retrieve it retrieved_token = token_cache.get_token("key1") assert token == retrieved_token + + +def test_unable_to_remove_cache(token_cache, tmp_path, monkeypatch, caplog): + def unlink_failure(*args): + raise PermissionError("You can create but can't delete") + + monkeypatch.setattr(pathlib.Path, "unlink", unlink_failure) + + caplog.set_level(logging.DEBUG) + # Assert that the exception is not thrown + token_cache.clear() + + assert caplog.messages == ["Clearing token cache.", "Cannot remove tokens file."] diff --git a/tests/test_oauth2_implicit.py b/tests/test_oauth2_implicit.py index c8301ea..0aa1cb5 100644 --- a/tests/test_oauth2_implicit.py +++ b/tests/test_oauth2_implicit.py @@ -34,7 +34,9 @@ def test_oauth2_implicit_flow_token_is_not_reused_if_a_url_parameter_is_changing "http://provide_token?response_type=custom_token&fake_param=1", token_field_name="custom_token", ) - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) first_token = create_token(expiry_in_1_hour) tab1 = browser_mock.add_response( opened_url="http://provide_token?response_type=custom_token&fake_param=1&state=5652a8138e3a99dab7b94532c73ed5b10f19405316035d1efdc8bf7e0713690485254c2eaff912040eac44031889ef0a5ed5730c8a111541120d64a898c31afe&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F", @@ -45,9 +47,9 @@ def test_oauth2_implicit_flow_token_is_not_reused_if_a_url_parameter_is_changing assert get_header(responses, auth1).get("Authorization") == f"Bearer {first_token}" # Ensure that the new token is different than previous one - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta( - hours=1, seconds=1 - ) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1, seconds=1) auth2 = requests_auth.OAuth2Implicit( "http://provide_token?response_type=custom_token&fake_param=2", @@ -77,7 +79,9 @@ def test_oauth2_implicit_flow_token_is_reused_if_only_nonce_differs( "http://provide_token?response_type=custom_token&nonce=1", token_field_name="custom_token", ) - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token = create_token(expiry_in_1_hour) tab = browser_mock.add_response( opened_url="http://provide_token?response_type=custom_token&state=67b95d2c7555751d1d72c97c7cd9ad6630c8395e0eaa51ee86ac7e451211ded9cd98a7190848789fe93632d8960425710e93f1f5549c6c6bc328bf3865a85ff2&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F&nonce=%5B%271%27%5D", @@ -106,7 +110,9 @@ def test_oauth2_implicit_flow_token_can_be_requested_on_a_custom_server_port( auth = requests_auth.OAuth2Implicit( "http://provide_token", redirect_uri_port=available_port ) - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token = create_token(expiry_in_1_hour) tab = browser_mock.add_response( opened_url="http://provide_token?response_type=token&state=42a85b271b7a652ca3cc4c398cfd3f01b9ad36bf9c945ba823b023e8f8b95c4638576a0e3dcc96838b838bec33ec6c0ee2609d62ed82480b3b8114ca494c0521&redirect_uri=http%3A%2F%2Flocalhost%3A5002%2F", @@ -123,7 +129,9 @@ def test_oauth2_implicit_flow_post_token_is_sent_in_authorization_header_by_defa token_cache, responses: RequestsMock, browser_mock: BrowserMock ): auth = requests_auth.OAuth2Implicit("http://provide_token") - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token = create_token(expiry_in_1_hour) tab = browser_mock.add_response( opened_url="http://provide_token?response_type=token&state=42a85b271b7a652ca3cc4c398cfd3f01b9ad36bf9c945ba823b023e8f8b95c4638576a0e3dcc96838b838bec33ec6c0ee2609d62ed82480b3b8114ca494c0521&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F", @@ -141,14 +149,18 @@ def test_oauth2_implicit_flow_token_is_expired_after_30_seconds_by_default( ): auth = requests_auth.OAuth2Implicit("http://provide_token") # Add a token that expires in 29 seconds, so should be considered as expired when issuing the request - expiry_in_29_seconds = datetime.datetime.utcnow() + datetime.timedelta(seconds=29) + expiry_in_29_seconds = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(seconds=29) token_cache._add_token( key="42a85b271b7a652ca3cc4c398cfd3f01b9ad36bf9c945ba823b023e8f8b95c4638576a0e3dcc96838b838bec33ec6c0ee2609d62ed82480b3b8114ca494c0521", token=create_token(expiry_in_29_seconds), expiry=requests_auth._oauth2.tokens._to_expiry(expires_in=29), ) # Meaning a new one will be requested - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token = create_token(expiry_in_1_hour) tab = browser_mock.add_response( opened_url="http://provide_token?response_type=token&state=42a85b271b7a652ca3cc4c398cfd3f01b9ad36bf9c945ba823b023e8f8b95c4638576a0e3dcc96838b838bec33ec6c0ee2609d62ed82480b3b8114ca494c0521&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F", @@ -166,7 +178,9 @@ def test_oauth2_implicit_flow_token_custom_expiry( ): auth = requests_auth.OAuth2Implicit("http://provide_token", early_expiry=28) # Add a token that expires in 29 seconds, so should be considered as not expired when issuing the request - expiry_in_29_seconds = datetime.datetime.utcnow() + datetime.timedelta(seconds=29) + expiry_in_29_seconds = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(seconds=29) token_cache._add_token( key="42a85b271b7a652ca3cc4c398cfd3f01b9ad36bf9c945ba823b023e8f8b95c4638576a0e3dcc96838b838bec33ec6c0ee2609d62ed82480b3b8114ca494c0521", token=create_token(expiry_in_29_seconds), @@ -234,7 +248,9 @@ def open(self, url, new): def test_state_change(token_cache, responses: RequestsMock, browser_mock: BrowserMock): auth = requests_auth.OAuth2Implicit("http://provide_token") - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token = create_token(expiry_in_1_hour) tab = browser_mock.add_response( opened_url="http://provide_token?response_type=token&state=42a85b271b7a652ca3cc4c398cfd3f01b9ad36bf9c945ba823b023e8f8b95c4638576a0e3dcc96838b838bec33ec6c0ee2609d62ed82480b3b8114ca494c0521&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F", @@ -283,7 +299,9 @@ def test_oauth2_implicit_flow_get_token_is_sent_in_authorization_header_by_defau token_cache, responses: RequestsMock, browser_mock: BrowserMock ): auth = requests_auth.OAuth2Implicit("http://provide_token") - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token = create_token(expiry_in_1_hour) tab = browser_mock.add_response( opened_url="http://provide_token?response_type=token&state=42a85b271b7a652ca3cc4c398cfd3f01b9ad36bf9c945ba823b023e8f8b95c4638576a0e3dcc96838b838bec33ec6c0ee2609d62ed82480b3b8114ca494c0521&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F", @@ -301,7 +319,9 @@ def test_oauth2_implicit_flow_token_is_sent_in_requested_field( auth = requests_auth.OAuth2Implicit( "http://provide_token", header_name="Bearer", header_value="{token}" ) - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token = create_token(expiry_in_1_hour) tab = browser_mock.add_response( opened_url="http://provide_token?response_type=token&state=42a85b271b7a652ca3cc4c398cfd3f01b9ad36bf9c945ba823b023e8f8b95c4638576a0e3dcc96838b838bec33ec6c0ee2609d62ed82480b3b8114ca494c0521&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F", @@ -322,7 +342,9 @@ def test_oauth2_implicit_flow_can_send_a_custom_response_type_and_expects_token_ response_type="custom_token", token_field_name="custom_token", ) - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token = create_token(expiry_in_1_hour) tab = browser_mock.add_response( opened_url="http://provide_token?response_type=custom_token&state=67b95d2c7555751d1d72c97c7cd9ad6630c8395e0eaa51ee86ac7e451211ded9cd98a7190848789fe93632d8960425710e93f1f5549c6c6bc328bf3865a85ff2&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F", @@ -341,7 +363,9 @@ def test_oauth2_implicit_flow_expects_token_in_id_token_if_response_type_is_id_t auth = requests_auth.OAuth2Implicit( "http://provide_token", response_type="id_token" ) - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token = create_token(expiry_in_1_hour) tab = browser_mock.add_response( opened_url="http://provide_token?response_type=id_token&state=87c4108ec0eb03599335333a40434a36674269690b6957fef684bfb6c5a849ce660ef7031aa874c44d67cd3eada8febdfce41efb1ed3bc53a0a7e716cbba025a&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F", @@ -358,7 +382,9 @@ def test_oauth2_implicit_flow_expects_token_in_id_token_if_response_type_in_url_ token_cache, responses: RequestsMock, browser_mock: BrowserMock ): auth = requests_auth.OAuth2Implicit("http://provide_token?response_type=id_token") - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token = create_token(expiry_in_1_hour) tab = browser_mock.add_response( opened_url="http://provide_token?response_type=id_token&state=87c4108ec0eb03599335333a40434a36674269690b6957fef684bfb6c5a849ce660ef7031aa874c44d67cd3eada8febdfce41efb1ed3bc53a0a7e716cbba025a&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F", @@ -375,7 +401,9 @@ def test_oauth2_implicit_flow_expects_token_to_be_stored_in_access_token_by_defa token_cache, responses: RequestsMock, browser_mock: BrowserMock ): auth = requests_auth.OAuth2Implicit("http://provide_token") - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token = create_token(expiry_in_1_hour) tab = browser_mock.add_response( opened_url="http://provide_token?response_type=token&state=42a85b271b7a652ca3cc4c398cfd3f01b9ad36bf9c945ba823b023e8f8b95c4638576a0e3dcc96838b838bec33ec6c0ee2609d62ed82480b3b8114ca494c0521&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F", @@ -392,7 +420,9 @@ def test_oauth2_implicit_flow_token_is_reused_if_not_expired( token_cache, responses: RequestsMock, browser_mock: BrowserMock ): auth1 = requests_auth.OAuth2Implicit("http://provide_token") - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token = create_token(expiry_in_1_hour) tab = browser_mock.add_response( opened_url="http://provide_token?response_type=token&state=42a85b271b7a652ca3cc4c398cfd3f01b9ad36bf9c945ba823b023e8f8b95c4638576a0e3dcc96838b838bec33ec6c0ee2609d62ed82480b3b8114ca494c0521&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F", @@ -450,7 +480,9 @@ def test_oauth2_implicit_flow_get_failure_if_token_is_not_provided( def test_oauth2_implicit_flow_post_failure_if_state_is_not_provided( token_cache, browser_mock: BrowserMock ): - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token = create_token(expiry_in_1_hour) tab = browser_mock.add_response( opened_url="http://provide_token?response_type=token&state=42a85b271b7a652ca3cc4c398cfd3f01b9ad36bf9c945ba823b023e8f8b95c4638576a0e3dcc96838b838bec33ec6c0ee2609d62ed82480b3b8114ca494c0521&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F", @@ -474,7 +506,9 @@ def test_oauth2_implicit_flow_post_failure_if_state_is_not_provided( def test_oauth2_implicit_flow_get_failure_if_state_is_not_provided( token_cache, browser_mock: BrowserMock ): - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) token = create_token(expiry_in_1_hour) tab = browser_mock.add_response( opened_url="http://provide_token?response_type=token&state=42a85b271b7a652ca3cc4c398cfd3f01b9ad36bf9c945ba823b023e8f8b95c4638576a0e3dcc96838b838bec33ec6c0ee2609d62ed82480b3b8114ca494c0521&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F", @@ -725,9 +759,9 @@ def test_oauth2_implicit_flow_token_is_requested_again_if_expired( ): auth = requests_auth.OAuth2Implicit("http://provide_token") # This token will expires in 100 milliseconds - expiry_in_1_second = datetime.datetime.utcnow() + datetime.timedelta( - milliseconds=100 - ) + expiry_in_1_second = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(milliseconds=100) first_token = create_token(expiry_in_1_second) tab1 = browser_mock.add_response( opened_url="http://provide_token?response_type=token&state=42a85b271b7a652ca3cc4c398cfd3f01b9ad36bf9c945ba823b023e8f8b95c4638576a0e3dcc96838b838bec33ec6c0ee2609d62ed82480b3b8114ca494c0521&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F", @@ -740,7 +774,9 @@ def test_oauth2_implicit_flow_token_is_requested_again_if_expired( time.sleep(0.2) # Token should now be expired, a new one should be requested - expiry_in_1_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) + expiry_in_1_hour = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(hours=1) second_token = create_token(expiry_in_1_hour) tab2 = browser_mock.add_response( opened_url="http://provide_token?response_type=token&state=42a85b271b7a652ca3cc4c398cfd3f01b9ad36bf9c945ba823b023e8f8b95c4638576a0e3dcc96838b838bec33ec6c0ee2609d62ed82480b3b8114ca494c0521&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F",