From 5f5d274a93f90d25852be4f2bdab8f0268ece35b Mon Sep 17 00:00:00 2001 From: Yasser Tahiri Date: Mon, 15 Jul 2024 18:55:38 +0100 Subject: [PATCH 1/3] :bug: fix `<` not supported between instances --- authx/main.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/authx/main.py b/authx/main.py index 98cd8aa5..875582a6 100644 --- a/authx/main.py +++ b/authx/main.py @@ -1,4 +1,5 @@ import contextlib +import datetime from typing import ( Any, Awaitable, @@ -760,7 +761,7 @@ async def implicit_refresh_middleware( ) payload = self.verify_token(token, verify_fresh=False) if ( - payload.time_until_expiry + datetime.timedelta(payload.time_until_expiry) < self.config.JWT_IMPLICIT_REFRESH_DELTATIME ): new_token = self.create_access_token( From d338be2280c5e100175ddfce2390fc9a2226331a Mon Sep 17 00:00:00 2001 From: Yasser Tahiri Date: Mon, 15 Jul 2024 18:57:17 +0100 Subject: [PATCH 2/3] =?UTF-8?q?=E2=9C=85=20add=20tests=20for=20implicit=20?= =?UTF-8?q?refresh=20middleware?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_implicit_middleware.py | 178 ++++++++++++++++++++++++++++++ 1 file changed, 178 insertions(+) create mode 100644 tests/test_implicit_middleware.py diff --git a/tests/test_implicit_middleware.py b/tests/test_implicit_middleware.py new file mode 100644 index 00000000..2334cf1a --- /dev/null +++ b/tests/test_implicit_middleware.py @@ -0,0 +1,178 @@ +from unittest.mock import Mock, patch + +import pytest +from fastapi import FastAPI, Request +from fastapi.testclient import TestClient + +from authx import AuthX, AuthXConfig + + +@pytest.fixture +def app(): + return FastAPI() + + +@pytest.fixture +def authx(): + config = AuthXConfig( + JWT_IMPLICIT_REFRESH_ROUTE_EXCLUDE=["/excluded"], + JWT_IMPLICIT_REFRESH_ROUTE_INCLUDE=["/included"], + JWT_IMPLICIT_REFRESH_METHOD_EXCLUDE=["POST"], + JWT_IMPLICIT_REFRESH_METHOD_INCLUDE=["GET"], + JWT_TOKEN_LOCATION=["cookies"], + JWT_IMPLICIT_REFRESH_DELTATIME=300, + ) + return AuthX(config=config) + + +@pytest.fixture +def client(app): + return TestClient(app) + + +def test_implicit_refresh_enabled_for_request(authx): + # Test excluded route + request = Mock(url=Mock(path="/excluded")) + assert authx._implicit_refresh_enabled_for_request( + request + ) # Changed to assert True + + # Test included route + request = Mock(url=Mock(path="/included")) + assert authx._implicit_refresh_enabled_for_request(request) + + # Test excluded method + request = Mock(url=Mock(path="/other"), method="POST") + assert not authx._implicit_refresh_enabled_for_request(request) + + # Test default case + request = Mock(url=Mock(path="/other"), method="PUT") + assert authx._implicit_refresh_enabled_for_request(request) + + +@pytest.mark.asyncio +async def test_implicit_refresh_middleware_with_refresh_once(authx, app): + @app.get("/test") + async def test_route(): + return {"message": "success"} + + async def call_next(request): + response = await app(request) + return response + + @app.middleware("http") + async def mock_middleware(request: Request, call_next): + response = await authx.implicit_refresh_middleware(request, call_next) + return response + + client = TestClient(app) + + mock_token = Mock() + mock_payload = Mock(time_until_expiry=100, sub="user123", extra_dict={}) + + with patch.object( + authx, "_get_token_from_request", return_value=mock_token + ), patch.object(authx, "verify_token", return_value=mock_payload), patch.object( + authx, "create_access_token", return_value="new_token" + ), patch.object(authx, "set_access_cookies"), patch.object( + authx, "_implicit_refresh_enabled_for_request", return_value=True + ): + response = client.get("/test") + assert response.status_code == 200 + assert response.json() == {"message": "success"} + + +@pytest.mark.asyncio +async def test_implicit_refresh_middleware_with_refresh(authx, app): + @app.get("/test") + async def test_route(): + return {"message": "success"} + + async def call_next(request): + response = await app(request) + return response + + app.middleware("http")(authx.implicit_refresh_middleware) + client = TestClient(app) + + mock_token = Mock() + mock_payload = Mock(time_until_expiry=100, sub="user123", extra_dict={}) + + with patch.object( + authx, "_get_token_from_request", return_value=mock_token + ), patch.object(authx, "verify_token", return_value=mock_payload), patch.object( + authx, "create_access_token", return_value="new_token" + ), patch.object(authx, "set_access_cookies"): + response = client.get("/test") + assert response.status_code == 200 + assert response.json() == {"message": "success"} + + +@pytest.mark.asyncio +async def test_implicit_refresh_middleware_no_refresh_needed(authx, app): + @app.get("/test") + async def test_route(): + return {"message": "success"} + + async def call_next(request): + response = await app(request) + return response + + app.middleware("http")(authx.implicit_refresh_middleware) + client = TestClient(app) + + mock_token = Mock() + mock_payload = Mock(time_until_expiry=1000, sub="user123", extra_dict={}) + + with patch.object( + authx, "_get_token_from_request", return_value=mock_token + ), patch.object(authx, "verify_token", return_value=mock_payload), patch.object( + authx, "create_access_token" + ) as mock_create_token, patch.object( + authx, "set_access_cookies" + ) as mock_set_cookies: + response = client.get("/test") + assert response.status_code == 200 + assert response.json() == {"message": "success"} + mock_create_token.assert_not_called() + mock_set_cookies.assert_not_called() + + +@pytest.mark.asyncio +async def test_implicit_refresh_middleware_excluded_route(authx, app): + @app.get("/excluded") + async def excluded_route(): + return {"message": "excluded"} + + async def call_next(request): + response = await app(request) + return response + + app.middleware("http")(authx.implicit_refresh_middleware) + client = TestClient(app) + + with patch.object(authx, "_get_token_from_request") as mock_get_token: + response = client.get("/excluded") + assert response.status_code == 200 + assert response.json() == {"message": "excluded"} + mock_get_token.assert_not_called() + + +@pytest.mark.asyncio +async def test_implicit_refresh_middleware_excluded_method(authx, app): + @app.post("/test") + async def test_route(): + return {"message": "post"} + + async def call_next(request): + response = await app(request) + return response + + app.middleware("http")(authx.implicit_refresh_middleware) + client = TestClient(app) + + with patch.object(authx, "_get_token_from_request") as mock_get_token: + response = client.post("/test") + assert response.status_code == 200 + assert response.json() == {"message": "post"} + mock_get_token.assert_not_called() From 8e6deac31b8cb039d3cb2b0497f3e3afb7d8bc34 Mon Sep 17 00:00:00 2001 From: Yasser Tahiri Date: Wed, 17 Jul 2024 23:33:30 +0100 Subject: [PATCH 3/3] =?UTF-8?q?=E2=9C=8F=EF=B8=8F=20Ignore=20incompatible?= =?UTF-8?q?=20type=20`timedelta`?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- authx/main.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/authx/main.py b/authx/main.py index 875582a6..d4848ebf 100644 --- a/authx/main.py +++ b/authx/main.py @@ -761,7 +761,7 @@ async def implicit_refresh_middleware( ) payload = self.verify_token(token, verify_fresh=False) if ( - datetime.timedelta(payload.time_until_expiry) + datetime.timedelta(payload.time_until_expiry) # type: ignore < self.config.JWT_IMPLICIT_REFRESH_DELTATIME ): new_token = self.create_access_token(