python/fastapi-users
1
0
Fork 0
mirror of https://github.com/fastapi-users/fastapi-users.git synced 2025-08-14 18:58:10 +08:00
Code Issues Packages Projects Releases Wiki Activity

Update for httpx-oauth >= 0.10 where account_email can be None

Browse Source
This commit is contained in:
François Voron
2022-11-04 09:35:51 +01:00
parent 08860167d1
commit c91e7657db
4 changed files with 69 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
fastapi_users/router/common.py
View File

@ -16,6 +16,7 @@ class ErrorCodeReasonModel(BaseModel):
class ErrorCode(str, Enum):
REGISTER_INVALID_PASSWORD = "REGISTER_INVALID_PASSWORD"
REGISTER_USER_ALREADY_EXISTS = "REGISTER_USER_ALREADY_EXISTS"
OAUTH_NOT_AVAILABLE_EMAIL = "OAUTH_NOT_AVAILABLE_EMAIL"
OAUTH_USER_ALREADY_EXISTS = "OAUTH_USER_ALREADY_EXISTS"
LOGIN_BAD_CREDENTIALS = "LOGIN_BAD_CREDENTIALS"
LOGIN_USER_NOT_VERIFIED = "LOGIN_USER_NOT_VERIFIED"

12
fastapi_users/router/oauth.py
View File

@ -111,6 +111,12 @@ def get_oauth_router(
token["access_token"]
)
if account_email is None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=ErrorCode.OAUTH_NOT_AVAILABLE_EMAIL,
)
try:
decode_jwt(state, state_secret, [STATE_TOKEN_AUDIENCE])
except jwt.DecodeError:
@ -235,6 +241,12 @@ def get_oauth_associate_router(
token["access_token"]
)
if account_email is None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=ErrorCode.OAUTH_NOT_AVAILABLE_EMAIL,
)
try:
state_data = decode_jwt(state, state_secret, [STATE_TOKEN_AUDIENCE])
except jwt.DecodeError:

2
pyproject.toml
View File

@ -141,7 +141,7 @@ beanie = [
"fastapi-users-db-beanie >=1.0.0",
]
oauth = [
"httpx-oauth >=0.4,<0.8"
"httpx-oauth >=0.4,<0.11"
]
redis = [
"redis >=4.3.3,<5.0.0",

55
tests/test_router_oauth.py
View File

@ -283,6 +283,33 @@ class TestCallback:
assert data["access_token"] == str(user_oauth.id)
assert user_manager_oauth.on_after_login.called is True
async def test_email_not_available(
self,
async_method_mocker: AsyncMethodMocker,
test_app_client_redirect_url: httpx.AsyncClient,
oauth_client: BaseOAuth2,
user_oauth: UserOAuthModel,
user_manager_oauth: UserManagerMock,
access_token: str,
):
state_jwt = generate_state_token({}, "SECRET")
async_method_mocker(oauth_client, "get_access_token", return_value=access_token)
async_method_mocker(
oauth_client, "get_id_email", return_value=("user_oauth1", None)
)
async_method_mocker(
user_manager_oauth, "oauth_callback", return_value=user_oauth
)
response = await test_app_client_redirect_url.get(
"/oauth/callback",
params={"code": "CODE", "state": state_jwt},
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
json = response.json()
assert json["detail"] == ErrorCode.OAUTH_NOT_AVAILABLE_EMAIL
@pytest.mark.router
@pytest.mark.oauth
@ -498,6 +525,34 @@ class TestAssociateCallback:
data = cast(Dict[str, Any], response.json())
assert data["id"] == str(user_oauth.id)
async def test_not_available_email(
self,
async_method_mocker: AsyncMethodMocker,
test_app_client_redirect_url: httpx.AsyncClient,
oauth_client: BaseOAuth2,
user_oauth: UserOAuthModel,
user_manager_oauth: UserManagerMock,
access_token: str,
):
state_jwt = generate_state_token({"sub": str(user_oauth.id)}, "SECRET")
async_method_mocker(oauth_client, "get_access_token", return_value=access_token)
async_method_mocker(
oauth_client, "get_id_email", return_value=("user_oauth1", None)
)
async_method_mocker(
user_manager_oauth, "oauth_callback", return_value=user_oauth
)
response = await test_app_client_redirect_url.get(
"/oauth-associate/callback",
params={"code": "CODE", "state": state_jwt},
headers={"Authorization": f"Bearer {user_oauth.id}"},
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
json = response.json()
assert json["detail"] == ErrorCode.OAUTH_NOT_AVAILABLE_EMAIL
@pytest.mark.asyncio
@pytest.mark.oauth