From bfc31efd6cb4a984d10b56e7bb7bb0a4609c9f8a Mon Sep 17 00:00:00 2001 From: Eshaan Bansal Date: Thu, 9 Nov 2023 16:45:34 +0530 Subject: [PATCH 01/13] feat: add new backend BitbucketDataCenterOAuth2 --- social_core/backends/bitbucket_datacenter.py | 126 +++++++++++++++++++ 1 file changed, 126 insertions(+) create mode 100644 social_core/backends/bitbucket_datacenter.py diff --git a/social_core/backends/bitbucket_datacenter.py b/social_core/backends/bitbucket_datacenter.py new file mode 100644 index 000000000..b0437cdbf --- /dev/null +++ b/social_core/backends/bitbucket_datacenter.py @@ -0,0 +1,126 @@ +import base64 +import hashlib + +from social_core.backends.oauth import BaseOAuth2 +from social_core.exceptions import AuthException + + +class BitbucketDataCenterOAuth2(BaseOAuth2): + """ + Implements client for Bitbucket Data Center OAuth 2.0 provider API. + ref: https://confluence.atlassian.com/bitbucketserver/bitbucket-oauth-2-0-provider-api-1108483661.html + """ + + name = "bitbucket-datacenter-oauth2" + ID_KEY = "id" + SCOPE_SEPARATOR = " " + ACCESS_TOKEN_METHOD = "POST" + REFRESH_TOKEN_METHOD = "POST" + REDIRECT_STATE = False + STATE_PARAMETER = True + USE_BASIC_AUTH = False + EXTRA_DATA = [ + ("token_type", "token_type"), + ("access_token", "access_token"), + ("refresh_token", "refresh_token"), + ("expires_in", "expires_in"), + ("scope", "scope"), + # extra user profile fields + ("name", "name"), + ("username", "username"), + ("display_name", "display_name"), + ("type", "type"), + ("active", "active"), + ("url", "url"), + ] + PKCE_DEFAULT_CODE_CHALLENGE_METHOD = "s256" + + @property + def server_base_oauth2_api_url(self): + base_url = self.setting("URL") + return f"{base_url}/rest/oauth2/latest" + + @property + def server_base_rest_api_url(self): + base_url = self.setting("URL") + return f"{base_url}/rest/api/latest" + + def authorization_url(self): + return f"{self.server_base_oauth2_api_url}/authorize" + + def access_token_url(self): + return f"{self.server_base_oauth2_api_url}/token" + + def get_user_details(self, response): + """Return user details for the Bitbucket Data Center account""" + # `response` here is the return value of `user_data` method + user_data = response + _, first_name, last_name = self.get_user_names(user_data["displayName"]) + uid = self.get_user_id(details=None, response=response) + return { + "uid": uid, + "first_name": first_name, + "last_name": last_name, + "email": user_data["emailAddress"], + "name": user_data["name"], + "username": user_data["slug"], + "display_name": user_data["displayName"], + "type": user_data["type"], + "active": user_data["active"], + "url": user_data["links"]["self"][0]["href"], + } + + def user_data(self, access_token, *args, **kwargs): + """Fetch user data from Bitbucket Data Center REST API""" + # ref: https://developer.atlassian.com/server/bitbucket/rest/v815/api-group-system-maintenance/#api-api-latest-users-get + response = self.get_json( + f"{self.server_base_rest_api_url}/users", + headers={"Authorization": f"Bearer {access_token}"}, + ) + return response["values"][0] + + def create_code_verifier(self): + name = self.name + "_code_verifier" + code_verifier = self.strategy.random_string(48) + self.strategy.session_set(name, code_verifier) + return code_verifier + + def get_code_verifier(self): + name = self.name + "_code_verifier" + code_verifier = self.strategy.session_get(name) + return code_verifier + + def generate_code_challenge(self, code_verifier, challenge_method): + method = challenge_method.lower() + if method == "s256": + hashed = hashlib.sha256(code_verifier.encode()).digest() + encoded = base64.urlsafe_b64encode(hashed) + code_challenge = encoded.decode().replace("=", "") # remove padding + return code_challenge + elif method == "plain": + return code_verifier + else: + raise AuthException("Unsupported code challenge method.") + + def auth_params(self, state=None): + params = super().auth_params(state=state) + + code_challenge_method = self.setting( + "PKCE_CODE_CHALLENGE_METHOD", + default=self.PKCE_DEFAULT_CODE_CHALLENGE_METHOD, + ) + code_verifier = self.create_code_verifier() + code_challenge = self.generate_code_challenge( + code_verifier, code_challenge_method + ) + params["code_challenge_method"] = code_challenge_method + params["code_challenge"] = code_challenge + return params + + def auth_complete_params(self, state=None): + params = super().auth_complete_params(state=state) + + code_verifier = self.get_code_verifier() + params["code_verifier"] = code_verifier + + return params From a3c6546893723a0bf626017631c18398ffcd6926 Mon Sep 17 00:00:00 2001 From: Eshaan Bansal Date: Thu, 9 Nov 2023 17:24:41 +0530 Subject: [PATCH 02/13] abstract away PKCE logic in BaseOAuth2PKCE for reuse --- social_core/backends/bitbucket_datacenter.py | 58 ++-------------- social_core/backends/oauth.py | 69 ++++++++++++++++++++ social_core/backends/twitter_oauth2.py | 56 +--------------- 3 files changed, 77 insertions(+), 106 deletions(-) diff --git a/social_core/backends/bitbucket_datacenter.py b/social_core/backends/bitbucket_datacenter.py index b0437cdbf..cd5e5a572 100644 --- a/social_core/backends/bitbucket_datacenter.py +++ b/social_core/backends/bitbucket_datacenter.py @@ -1,11 +1,7 @@ -import base64 -import hashlib +from social_core.backends.oauth import BaseOAuth2PKCE -from social_core.backends.oauth import BaseOAuth2 -from social_core.exceptions import AuthException - -class BitbucketDataCenterOAuth2(BaseOAuth2): +class BitbucketDataCenterOAuth2(BaseOAuth2PKCE): """ Implements client for Bitbucket Data Center OAuth 2.0 provider API. ref: https://confluence.atlassian.com/bitbucketserver/bitbucket-oauth-2-0-provider-api-1108483661.html @@ -33,7 +29,9 @@ class BitbucketDataCenterOAuth2(BaseOAuth2): ("active", "active"), ("url", "url"), ] - PKCE_DEFAULT_CODE_CHALLENGE_METHOD = "s256" + PKCE_DEFAULT_CODE_CHALLENGE_METHOD = "s256" # can be "plain" or "s256" + PKCE_DEFAULT_CODE_VERIFIER_LENGTH = 48 # must be b/w 43-127 chars + USE_PKCE = True @property def server_base_oauth2_api_url(self): @@ -78,49 +76,3 @@ def user_data(self, access_token, *args, **kwargs): headers={"Authorization": f"Bearer {access_token}"}, ) return response["values"][0] - - def create_code_verifier(self): - name = self.name + "_code_verifier" - code_verifier = self.strategy.random_string(48) - self.strategy.session_set(name, code_verifier) - return code_verifier - - def get_code_verifier(self): - name = self.name + "_code_verifier" - code_verifier = self.strategy.session_get(name) - return code_verifier - - def generate_code_challenge(self, code_verifier, challenge_method): - method = challenge_method.lower() - if method == "s256": - hashed = hashlib.sha256(code_verifier.encode()).digest() - encoded = base64.urlsafe_b64encode(hashed) - code_challenge = encoded.decode().replace("=", "") # remove padding - return code_challenge - elif method == "plain": - return code_verifier - else: - raise AuthException("Unsupported code challenge method.") - - def auth_params(self, state=None): - params = super().auth_params(state=state) - - code_challenge_method = self.setting( - "PKCE_CODE_CHALLENGE_METHOD", - default=self.PKCE_DEFAULT_CODE_CHALLENGE_METHOD, - ) - code_verifier = self.create_code_verifier() - code_challenge = self.generate_code_challenge( - code_verifier, code_challenge_method - ) - params["code_challenge_method"] = code_challenge_method - params["code_challenge"] = code_challenge - return params - - def auth_complete_params(self, state=None): - params = super().auth_complete_params(state=state) - - code_verifier = self.get_code_verifier() - params["code_verifier"] = code_verifier - - return params diff --git a/social_core/backends/oauth.py b/social_core/backends/oauth.py index ae00aa5b2..7476aef79 100644 --- a/social_core/backends/oauth.py +++ b/social_core/backends/oauth.py @@ -1,3 +1,5 @@ +import base64 +import hashlib from urllib.parse import unquote, urlencode from oauthlib.oauth1 import SIGNATURE_TYPE_AUTH_HEADER @@ -11,6 +13,7 @@ AuthStateMissing, AuthTokenError, AuthUnknownError, + AuthException, ) from ..utils import ( constant_time_compare, @@ -459,3 +462,69 @@ def refresh_token(self, token, *args, **kwargs): def refresh_token_url(self): return self.REFRESH_TOKEN_URL or self.access_token_url() + + +class BaseOAuth2PKCE(BaseOAuth2): + """ + Base class for providers using OAuth2 with Proof Key for Code Exchange (PKCE). + + OAuth2 details at: + https://datatracker.ietf.org/doc/html/rfc6749 + PKCE details at: + https://datatracker.ietf.org/doc/html/rfc7636 + """ + + PKCE_DEFAULT_CODE_CHALLENGE_METHOD = "s256" + PKCE_DEFAULT_CODE_VERIFIER_LENGTH = 32 + USE_PKCE = True + + def create_code_verifier(self): + name = self.name + "_code_verifier" + code_verifier_length = self.setting( + "CODE_VERIFIER_LENGTH", default=self.PKCE_DEFAULT_CODE_VERIFIER_LENGTH + ) + code_verifier = self.strategy.random_string(code_verifier_length) + self.strategy.session_set(name, code_verifier) + return code_verifier + + def get_code_verifier(self): + name = self.name + "_code_verifier" + code_verifier = self.strategy.session_get(name) + return code_verifier + + def generate_code_challenge(self, code_verifier, challenge_method): + method = challenge_method.lower() + if method == "s256": + hashed = hashlib.sha256(code_verifier.encode()).digest() + encoded = base64.urlsafe_b64encode(hashed) + code_challenge = encoded.decode().replace("=", "") # remove padding + return code_challenge + elif method == "plain": + return code_verifier + else: + raise AuthException("Unsupported code challenge method.") + + def auth_params(self, state=None): + params = super().auth_params(state=state) + + if self.USE_PKCE: + code_challenge_method = self.setting( + "PKCE_CODE_CHALLENGE_METHOD", + default=self.PKCE_DEFAULT_CODE_CHALLENGE_METHOD, + ) + code_verifier = self.create_code_verifier() + code_challenge = self.generate_code_challenge( + code_verifier, code_challenge_method + ) + params["code_challenge_method"] = code_challenge_method + params["code_challenge"] = code_challenge + return params + + def auth_complete_params(self, state=None): + params = super().auth_complete_params(state=state) + + if self.USE_PKCE: + code_verifier = self.get_code_verifier() + params["code_verifier"] = code_verifier + + return params diff --git a/social_core/backends/twitter_oauth2.py b/social_core/backends/twitter_oauth2.py index 4172fc36f..c56830dda 100644 --- a/social_core/backends/twitter_oauth2.py +++ b/social_core/backends/twitter_oauth2.py @@ -3,14 +3,10 @@ https://python-social-auth.readthedocs.io/en/latest/backends/twitter-oauth2.html https://developer.twitter.com/en/docs/authentication/oauth-2-0/authorization-code """ -import base64 -import hashlib +from .oauth import BaseOAuth2PKCE -from ..exceptions import AuthException -from .oauth import BaseOAuth2 - -class TwitterOAuth2(BaseOAuth2): +class TwitterOAuth2(BaseOAuth2PKCE): """Twitter OAuth2 authentication backend""" name = "twitter-oauth2" @@ -40,6 +36,7 @@ class TwitterOAuth2(BaseOAuth2): ("public_metrics", "public_metrics"), ] PKCE_DEFAULT_CODE_CHALLENGE_METHOD = "s256" + PKCE_DEFAULT_CODE_VERIFIER_LENGTH = 32 USE_PKCE = True def get_user_details(self, response): @@ -104,50 +101,3 @@ def user_data(self, access_token, *args, **kwargs): headers={"Authorization": "Bearer %s" % access_token}, ) return response["data"] - - def create_code_verifier(self): - name = self.name + "_code_verifier" - code_verifier = self.strategy.random_string(32) - self.strategy.session_set(name, code_verifier) - return code_verifier - - def get_code_verifier(self): - name = self.name + "_code_verifier" - code_verifier = self.strategy.session_get(name) - return code_verifier - - def generate_code_challenge(self, code_verifier, challenge_method): - method = challenge_method.lower() - if method == "s256": - hashed = hashlib.sha256(code_verifier.encode()).digest() - encoded = base64.urlsafe_b64encode(hashed) - code_challenge = encoded.decode().replace("=", "") # remove padding - return code_challenge - elif method == "plain": - return code_verifier - else: - raise AuthException("Unsupported code challenge method.") - - def auth_params(self, state=None): - params = super().auth_params(state=state) - - if self.USE_PKCE: - code_challenge_method = self.setting("PKCE_CODE_CHALLENGE_METHOD") - if not code_challenge_method: - code_challenge_method = self.PKCE_DEFAULT_CODE_CHALLENGE_METHOD - code_verifier = self.create_code_verifier() - code_challenge = self.generate_code_challenge( - code_verifier, code_challenge_method - ) - params["code_challenge_method"] = code_challenge_method - params["code_challenge"] = code_challenge - return params - - def auth_complete_params(self, state=None): - params = super().auth_complete_params(state=state) - - if self.USE_PKCE: - code_verifier = self.get_code_verifier() - params["code_verifier"] = code_verifier - - return params From 79685496d582175eb644969f85e89f685e63b941 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Thu, 9 Nov 2023 11:55:16 +0000 Subject: [PATCH 03/13] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- social_core/backends/oauth.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/social_core/backends/oauth.py b/social_core/backends/oauth.py index 7476aef79..9112dd04f 100644 --- a/social_core/backends/oauth.py +++ b/social_core/backends/oauth.py @@ -7,13 +7,13 @@ from ..exceptions import ( AuthCanceled, + AuthException, AuthFailed, AuthMissingParameter, AuthStateForbidden, AuthStateMissing, AuthTokenError, AuthUnknownError, - AuthException, ) from ..utils import ( constant_time_compare, From 2f0f8958d2d466a83e49cc959b5f2d5443e58dc9 Mon Sep 17 00:00:00 2001 From: Eshaan Bansal Date: Thu, 9 Nov 2023 17:29:36 +0530 Subject: [PATCH 04/13] abstract away PKCE logic in BaseOAuth2PKCE for reuse --- social_core/backends/oauth.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/social_core/backends/oauth.py b/social_core/backends/oauth.py index 9112dd04f..86f0afe8e 100644 --- a/social_core/backends/oauth.py +++ b/social_core/backends/oauth.py @@ -480,10 +480,10 @@ class BaseOAuth2PKCE(BaseOAuth2): def create_code_verifier(self): name = self.name + "_code_verifier" - code_verifier_length = self.setting( - "CODE_VERIFIER_LENGTH", default=self.PKCE_DEFAULT_CODE_VERIFIER_LENGTH + code_verifier_len = self.setting( + "PKCE_CODE_VERIFIER_LENGTH", default=self.PKCE_DEFAULT_CODE_VERIFIER_LENGTH ) - code_verifier = self.strategy.random_string(code_verifier_length) + code_verifier = self.strategy.random_string(code_verifier_len) self.strategy.session_set(name, code_verifier) return code_verifier From 72c1d7b0f4c1639d514800db9acdb2e7de8e70fc Mon Sep 17 00:00:00 2001 From: Eshaan Bansal Date: Thu, 9 Nov 2023 17:38:48 +0530 Subject: [PATCH 05/13] noqa flake8 line length rule for URLs in comments --- social_core/backends/bitbucket_datacenter.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/social_core/backends/bitbucket_datacenter.py b/social_core/backends/bitbucket_datacenter.py index cd5e5a572..08b047f64 100644 --- a/social_core/backends/bitbucket_datacenter.py +++ b/social_core/backends/bitbucket_datacenter.py @@ -14,6 +14,8 @@ class BitbucketDataCenterOAuth2(BaseOAuth2PKCE): REFRESH_TOKEN_METHOD = "POST" REDIRECT_STATE = False STATE_PARAMETER = True + # ref: https://confluence.atlassian.com/bitbucketserver/bitbucket-oauth-2-0-provider-api-1108483661.html#BitbucketOAuth2.0providerAPI-scopes # noqa + DEFAULT_SCOPE = ["PUBLIC_REPOS"] USE_BASIC_AUTH = False EXTRA_DATA = [ ("token_type", "token_type"), @@ -70,7 +72,7 @@ def get_user_details(self, response): def user_data(self, access_token, *args, **kwargs): """Fetch user data from Bitbucket Data Center REST API""" - # ref: https://developer.atlassian.com/server/bitbucket/rest/v815/api-group-system-maintenance/#api-api-latest-users-get + # ref: https://developer.atlassian.com/server/bitbucket/rest/v815/api-group-system-maintenance/#api-api-latest-users-get # noqa response = self.get_json( f"{self.server_base_rest_api_url}/users", headers={"Authorization": f"Bearer {access_token}"}, From 7c8bf02cae48c7999718af13d2845fef7ef9e761 Mon Sep 17 00:00:00 2001 From: Eshaan Bansal Date: Tue, 21 Nov 2023 14:57:58 +0530 Subject: [PATCH 06/13] chore: add BitbucketDataCenterOAuth2Test --- social_core/backends/bitbucket_datacenter.py | 51 +++++-- .../backends/test_bitbucket_datacenter.py | 137 ++++++++++++++++++ 2 files changed, 175 insertions(+), 13 deletions(-) create mode 100644 social_core/tests/backends/test_bitbucket_datacenter.py diff --git a/social_core/backends/bitbucket_datacenter.py b/social_core/backends/bitbucket_datacenter.py index 08b047f64..37d35968f 100644 --- a/social_core/backends/bitbucket_datacenter.py +++ b/social_core/backends/bitbucket_datacenter.py @@ -1,4 +1,10 @@ -from social_core.backends.oauth import BaseOAuth2PKCE +""" +Bitbucket Data Center OAuth2 backend, docs at: + https://python-social-auth.readthedocs.io/en/latest/backends/bitbucket-datacenter-oauth2.html + https://confluence.atlassian.com/bitbucketserver/bitbucket-oauth-2-0-provider-api-1108483661.html +""" + +from .oauth import BaseOAuth2PKCE class BitbucketDataCenterOAuth2(BaseOAuth2PKCE): @@ -21,37 +27,42 @@ class BitbucketDataCenterOAuth2(BaseOAuth2PKCE): ("token_type", "token_type"), ("access_token", "access_token"), ("refresh_token", "refresh_token"), - ("expires_in", "expires_in"), + ("expires_in", "expires"), ("scope", "scope"), # extra user profile fields + ("first_name", "first_name"), + ("last_name", "last_name"), + ("email", "email"), ("name", "name"), ("username", "username"), ("display_name", "display_name"), ("type", "type"), ("active", "active"), ("url", "url"), + ("avatar_url", "avatar_url"), ] PKCE_DEFAULT_CODE_CHALLENGE_METHOD = "s256" # can be "plain" or "s256" PKCE_DEFAULT_CODE_VERIFIER_LENGTH = 48 # must be b/w 43-127 chars USE_PKCE = True + DEFAULT_AVATAR_SIZE = 48 @property - def server_base_oauth2_api_url(self): + def server_base_oauth2_api_url(self) -> str: base_url = self.setting("URL") return f"{base_url}/rest/oauth2/latest" @property - def server_base_rest_api_url(self): + def server_base_rest_api_url(self) -> str: base_url = self.setting("URL") return f"{base_url}/rest/api/latest" - def authorization_url(self): + def authorization_url(self) -> str: return f"{self.server_base_oauth2_api_url}/authorize" - def access_token_url(self): + def access_token_url(self) -> str: return f"{self.server_base_oauth2_api_url}/token" - def get_user_details(self, response): + def get_user_details(self, response) -> dict: """Return user details for the Bitbucket Data Center account""" # `response` here is the return value of `user_data` method user_data = response @@ -68,13 +79,27 @@ def get_user_details(self, response): "type": user_data["type"], "active": user_data["active"], "url": user_data["links"]["self"][0]["href"], + "avatar_url": user_data["avatarUrl"], } - def user_data(self, access_token, *args, **kwargs): + def user_data(self, access_token, *args, **kwargs) -> dict: """Fetch user data from Bitbucket Data Center REST API""" - # ref: https://developer.atlassian.com/server/bitbucket/rest/v815/api-group-system-maintenance/#api-api-latest-users-get # noqa - response = self.get_json( - f"{self.server_base_rest_api_url}/users", - headers={"Authorization": f"Bearer {access_token}"}, + # At this point, we don't know the current user's username + # and Bitbucket doesn't provide any API to do so. + # However, the current user's username is sent in every response header. + # ref: https://community.developer.atlassian.com/t/obtain-authorised-users-username-from-api/24422/2 + headers = {"Authorization": f"Bearer {access_token}"} + response = self.request( + url=f"{self.server_base_rest_api_url}/application-properties", + method="GET", + headers=headers, + ) + # ref: https://developer.atlassian.com/server/bitbucket/rest/v815/api-group-system-maintenance/#api-api-latest-users-userslug-get + username = response.headers["x-ausername"] + return self.get_json( + url=f"{self.server_base_rest_api_url}/users/{username}", + headers=headers, + params={ + "avatarSize": self.DEFAULT_AVATAR_SIZE # to force `avatarUrl` in response + }, ) - return response["values"][0] diff --git a/social_core/tests/backends/test_bitbucket_datacenter.py b/social_core/tests/backends/test_bitbucket_datacenter.py new file mode 100644 index 000000000..6884c9824 --- /dev/null +++ b/social_core/tests/backends/test_bitbucket_datacenter.py @@ -0,0 +1,137 @@ +import json + +from httpretty import HTTPretty + +from .oauth import OAuth2Test + + +class BitbucketDataCenterOAuth2Test(OAuth2Test): + backend_path = "social_core.backends.bitbucket_datacenter.BitbucketDataCenterOAuth2" + application_properties_url = ( + "https://bachmanity.atlassian.net/rest/api/latest/application-properties" + ) + application_properties_headers = {"x-ausername": "erlich-bachman"} + application_properties_body = json.dumps( + { + "version": "8.15.0", + "buildNumber": "8015000", + "buildDate": "1697764661289", + "displayName": "Bitbucket", + } + ) + user_data_url = ( + "https://bachmanity.atlassian.net/rest/api/latest/users/erlich-bachman" + ) + user_data_body = json.dumps( + { + "name": "erlich-bachman", + "emailAddress": "erlich@bachmanity.com", + "active": True, + "displayName": "Erlich Bachman", + "id": 1, + "slug": "erlich-bachman", + "type": "NORMAL", + "links": { + "self": [ + {"href": "https://bachmanity.atlassian.net/users/erlich-bachman"} + ] + }, + "avatarUrl": "http://www.gravatar.com/avatar/af7d968fe79ea45271e3100391824b79.jpg?s=48&d=mm", + } + ) + access_token_body = json.dumps( + { + "scope": "PUBLIC_REPOS", + "access_token": "dummy_access_token", + "token_type": "bearer", + "expires_in": 3600, + "refresh_token": "dummy_refresh_token", + } + ) + refresh_token_body = json.dumps( + { + "scope": "PUBLIC_REPOS", + "access_token": "dummy_access_token_refreshed", + "token_type": "bearer", + "expires_in": 3600, + "refresh_token": "dummy_refresh_token_refreshed", + } + ) + expected_username = "erlich-bachman" + + def extra_settings(self): + settings = super().extra_settings() + settings.update( + {"SOCIAL_AUTH_" + self.name + "_URL": "https://bachmanity.atlassian.net"} + ) + return settings + + def auth_handlers(self, start_url): + target_url = super().auth_handlers(start_url) + HTTPretty.register_uri( + HTTPretty.GET, + self.application_properties_url, + body=self.application_properties_body, + adding_headers=self.application_properties_headers, + content_type="text/json", + ) + return target_url + + def test_login(self): + user = self.do_login() + + self.assertEqual(len(user.social), 1) + + social = user.social[0] + self.assertEqual(social.uid, 1) + self.assertEqual(social.extra_data["first_name"], "Erlich") + self.assertEqual(social.extra_data["last_name"], "Bachman") + self.assertEqual(social.extra_data["email"], "erlich@bachmanity.com") + self.assertEqual(social.extra_data["name"], "erlich-bachman") + self.assertEqual(social.extra_data["username"], "erlich-bachman") + self.assertEqual(social.extra_data["display_name"], "Erlich Bachman") + self.assertEqual(social.extra_data["type"], "NORMAL") + self.assertEqual(social.extra_data["active"], True) + self.assertEqual( + social.extra_data["url"], + "https://bachmanity.atlassian.net/users/erlich-bachman", + ) + self.assertEqual( + social.extra_data["avatar_url"], + "http://www.gravatar.com/avatar/af7d968fe79ea45271e3100391824b79.jpg?s=48&d=mm", + ) + self.assertEqual(social.extra_data["scope"], "PUBLIC_REPOS") + self.assertEqual(social.extra_data["access_token"], "dummy_access_token") + self.assertEqual(social.extra_data["token_type"], "bearer") + self.assertEqual(social.extra_data["expires"], 3600) + self.assertEqual(social.extra_data["refresh_token"], "dummy_refresh_token") + + def test_refresh_token(self): + _, social = self.do_refresh_token() + + self.assertEqual(social.uid, 1) + self.assertEqual(social.extra_data["first_name"], "Erlich") + self.assertEqual(social.extra_data["last_name"], "Bachman") + self.assertEqual(social.extra_data["email"], "erlich@bachmanity.com") + self.assertEqual(social.extra_data["name"], "erlich-bachman") + self.assertEqual(social.extra_data["username"], "erlich-bachman") + self.assertEqual(social.extra_data["display_name"], "Erlich Bachman") + self.assertEqual(social.extra_data["type"], "NORMAL") + self.assertEqual(social.extra_data["active"], True) + self.assertEqual( + social.extra_data["url"], + "https://bachmanity.atlassian.net/users/erlich-bachman", + ) + self.assertEqual( + social.extra_data["avatar_url"], + "http://www.gravatar.com/avatar/af7d968fe79ea45271e3100391824b79.jpg?s=48&d=mm", + ) + self.assertEqual(social.extra_data["scope"], "PUBLIC_REPOS") + self.assertEqual( + social.extra_data["access_token"], "dummy_access_token_refreshed" + ) + self.assertEqual(social.extra_data["token_type"], "bearer") + self.assertEqual(social.extra_data["expires"], 3600) + self.assertEqual( + social.extra_data["refresh_token"], "dummy_refresh_token_refreshed" + ) From 7277ec306240e11ae1e7d472c20de6df1dfc73b4 Mon Sep 17 00:00:00 2001 From: Eshaan Bansal Date: Tue, 21 Nov 2023 15:26:11 +0530 Subject: [PATCH 07/13] chore: abstract PKCE tests in OAuth2PkcePlainTest, OAuth2PkceS256Test --- social_core/tests/backends/oauth.py | 56 ++++++++++++++++- .../backends/test_bitbucket_datacenter.py | 18 +++++- .../tests/backends/test_twitter_oauth2.py | 61 ++++--------------- 3 files changed, 83 insertions(+), 52 deletions(-) diff --git a/social_core/tests/backends/oauth.py b/social_core/tests/backends/oauth.py index 44946d419..f8bb483cc 100644 --- a/social_core/tests/backends/oauth.py +++ b/social_core/tests/backends/oauth.py @@ -1,7 +1,7 @@ from urllib.parse import urlparse import requests -from httpretty import HTTPretty +from httpretty import HTTPretty, latest_requests from ...utils import parse_qs, url_add_parameters from ..models import User @@ -121,3 +121,57 @@ def do_refresh_token(self): social = user.social[0] social.refresh_token(strategy=self.strategy, **self.refresh_token_arguments()) return user, social + + +class OAuth2PkcePlainTest(OAuth2Test): + def extra_settings(self): + settings = super().extra_settings() + settings.update( + {"SOCIAL_AUTH_" + self.name + "_PKCE_CODE_CHALLENGE_METHOD": "plain"} + ) + return settings + + def do_login(self): + user = super().do_login() + + requests = latest_requests() + auth_request = [ + r for r in requests if self.backend.authorization_url() in r.url + ][0] + code_challenge = auth_request.querystring.get("code_challenge")[0] + code_challenge_method = auth_request.querystring.get("code_challenge_method")[0] + self.assertIsNotNone(code_challenge) + self.assertEqual(code_challenge_method, "plain") + + auth_complete = [ + r for r in requests if self.backend.access_token_url() in r.url + ][0] + code_verifier = auth_complete.parsed_body.get("code_verifier")[0] + self.assertEqual(code_challenge, code_verifier) + + return user + + +class OAuth2PkceS256Test(OAuth2Test): + def do_login(self): + # use default value of PKCE_CODE_CHALLENGE_METHOD (s256) + user = super().do_login() + + requests = latest_requests() + auth_request = [ + r for r in requests if self.backend.authorization_url() in r.url + ][0] + code_challenge = auth_request.querystring.get("code_challenge")[0] + code_challenge_method = auth_request.querystring.get("code_challenge_method")[0] + self.assertIsNotNone(code_challenge) + self.assertEqual(code_challenge_method, "s256") + + auth_complete = [ + r for r in requests if self.backend.access_token_url() in r.url + ][0] + code_verifier = auth_complete.parsed_body.get("code_verifier")[0] + self.assertEqual( + self.backend.generate_code_challenge(code_verifier, "s256"), code_challenge + ) + + return user diff --git a/social_core/tests/backends/test_bitbucket_datacenter.py b/social_core/tests/backends/test_bitbucket_datacenter.py index 6884c9824..aed0f07e7 100644 --- a/social_core/tests/backends/test_bitbucket_datacenter.py +++ b/social_core/tests/backends/test_bitbucket_datacenter.py @@ -2,10 +2,10 @@ from httpretty import HTTPretty -from .oauth import OAuth2Test +from .oauth import OAuth2PkcePlainTest, OAuth2PkceS256Test -class BitbucketDataCenterOAuth2Test(OAuth2Test): +class BitbucketDataCenterOAuth2Mixin: backend_path = "social_core.backends.bitbucket_datacenter.BitbucketDataCenterOAuth2" application_properties_url = ( "https://bachmanity.atlassian.net/rest/api/latest/application-properties" @@ -135,3 +135,17 @@ def test_refresh_token(self): self.assertEqual( social.extra_data["refresh_token"], "dummy_refresh_token_refreshed" ) + + +class BitbucketDataCenterOAuth2TestPkcePlain( + BitbucketDataCenterOAuth2Mixin, + OAuth2PkcePlainTest, +): + pass + + +class BitbucketDataCenterOAuth2TestPkceS256( + BitbucketDataCenterOAuth2Mixin, + OAuth2PkceS256Test, +): + pass diff --git a/social_core/tests/backends/test_twitter_oauth2.py b/social_core/tests/backends/test_twitter_oauth2.py index d003fb77f..fb799d170 100644 --- a/social_core/tests/backends/test_twitter_oauth2.py +++ b/social_core/tests/backends/test_twitter_oauth2.py @@ -1,13 +1,11 @@ import json -import httpretty - from social_core.exceptions import AuthException -from .oauth import OAuth2Test +from .oauth import OAuth2Test, OAuth2PkcePlainTest, OAuth2PkceS256Test -class TwitterOAuth2Test(OAuth2Test): +class TwitterOAuth2Mixin: backend_path = "social_core.backends.twitter_oauth2.TwitterOAuth2" user_data_url = "https://api.twitter.com/2/users/me" access_token_body = json.dumps( @@ -172,58 +170,23 @@ def test_login(self): self.assertIsNone(social.extra_data.get("public_metrics")) -class TwitterOAuth2TestPkcePlain(TwitterOAuth2Test): - def test_login(self): - self.strategy.set_settings( - {"SOCIAL_AUTH_TWITTER_OAUTH2_PKCE_CODE_CHALLENGE_METHOD": "plain"} - ) - - self.do_login() - - requests = httpretty.latest_requests() - auth_request = [ - r for r in requests if "https://twitter.com/i/oauth2/authorize" in r.url - ][0] - code_challenge = auth_request.querystring.get("code_challenge")[0] - code_challenge_method = auth_request.querystring.get("code_challenge_method")[0] - self.assertIsNotNone(code_challenge) - self.assertEqual(code_challenge_method, "plain") - - auth_complete = [ - r for r in requests if "https://api.twitter.com/2/oauth2/token" in r.url - ][0] - code_verifier = auth_complete.parsed_body.get("code_verifier")[0] - self.assertEqual(code_challenge, code_verifier) +class TwitterOAuth2TestPkcePlain(TwitterOAuth2Mixin, OAuth2PkcePlainTest): + pass -class TwitterOAuth2TestPkceS256(TwitterOAuth2Test): - def test_login(self): - # use default value of PKCE_CODE_CHALLENGE_METHOD (s256) - self.do_login() - - requests = httpretty.latest_requests() - auth_request = [ - r for r in requests if "https://twitter.com/i/oauth2/authorize" in r.url - ][0] - code_challenge = auth_request.querystring.get("code_challenge")[0] - code_challenge_method = auth_request.querystring.get("code_challenge_method")[0] - self.assertIsNotNone(code_challenge) - self.assertEqual(code_challenge_method, "s256") - - auth_complete = [ - r for r in requests if "https://api.twitter.com/2/oauth2/toke" in r.url - ][0] - code_verifier = auth_complete.parsed_body.get("code_verifier")[0] - self.assertEqual( - self.backend.generate_code_challenge(code_verifier, "s256"), code_challenge - ) +class TwitterOAuth2TestPkceS256(TwitterOAuth2Mixin, OAuth2PkceS256Test): + pass -class TwitterOAuth2TestInvalidCodeChallengeMethod(TwitterOAuth2Test): +class TwitterOAuth2TestInvalidCodeChallengeMethod( + TwitterOAuth2Mixin, OAuth2PkcePlainTest +): def test_login__error(self): self.strategy.set_settings( { - "SOCIAL_AUTH_TWITTER_OAUTH2_PKCE_CODE_CHALLENGE_METHOD": "invalidmethodname" + "SOCIAL_AUTH_" + + self.name + + "_PKCE_CODE_CHALLENGE_METHOD": "invalidmethodname" } ) From 0e50d2a1a626ffc92e9acab07fd8887e4b725b21 Mon Sep 17 00:00:00 2001 From: Eshaan Bansal Date: Tue, 21 Nov 2023 15:29:21 +0530 Subject: [PATCH 08/13] noqa flake8 line length rule for URLs in comments --- social_core/backends/bitbucket_datacenter.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/social_core/backends/bitbucket_datacenter.py b/social_core/backends/bitbucket_datacenter.py index 37d35968f..1569bb13e 100644 --- a/social_core/backends/bitbucket_datacenter.py +++ b/social_core/backends/bitbucket_datacenter.py @@ -87,14 +87,14 @@ def user_data(self, access_token, *args, **kwargs) -> dict: # At this point, we don't know the current user's username # and Bitbucket doesn't provide any API to do so. # However, the current user's username is sent in every response header. - # ref: https://community.developer.atlassian.com/t/obtain-authorised-users-username-from-api/24422/2 + # ref: https://community.developer.atlassian.com/t/obtain-authorised-users-username-from-api/24422/2 # noqa headers = {"Authorization": f"Bearer {access_token}"} response = self.request( url=f"{self.server_base_rest_api_url}/application-properties", method="GET", headers=headers, ) - # ref: https://developer.atlassian.com/server/bitbucket/rest/v815/api-group-system-maintenance/#api-api-latest-users-userslug-get + # ref: https://developer.atlassian.com/server/bitbucket/rest/v815/api-group-system-maintenance/#api-api-latest-users-userslug-get # noqa username = response.headers["x-ausername"] return self.get_json( url=f"{self.server_base_rest_api_url}/users/{username}", From 676441fe020415d099bf50a8129a04be4d18d4cc Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Tue, 21 Nov 2023 10:09:32 +0000 Subject: [PATCH 09/13] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- social_core/tests/backends/test_twitter_oauth2.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/social_core/tests/backends/test_twitter_oauth2.py b/social_core/tests/backends/test_twitter_oauth2.py index fb799d170..f54878958 100644 --- a/social_core/tests/backends/test_twitter_oauth2.py +++ b/social_core/tests/backends/test_twitter_oauth2.py @@ -2,7 +2,7 @@ from social_core.exceptions import AuthException -from .oauth import OAuth2Test, OAuth2PkcePlainTest, OAuth2PkceS256Test +from .oauth import OAuth2PkcePlainTest, OAuth2PkceS256Test, OAuth2Test class TwitterOAuth2Mixin: From a413849f23f31ca6a954e50c414179861373d6d2 Mon Sep 17 00:00:00 2001 From: Eshaan Bansal Date: Tue, 21 Nov 2023 15:41:15 +0530 Subject: [PATCH 10/13] chore: fix isort errors --- social_core/tests/backends/test_twitter_oauth2.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/social_core/tests/backends/test_twitter_oauth2.py b/social_core/tests/backends/test_twitter_oauth2.py index fb799d170..f54878958 100644 --- a/social_core/tests/backends/test_twitter_oauth2.py +++ b/social_core/tests/backends/test_twitter_oauth2.py @@ -2,7 +2,7 @@ from social_core.exceptions import AuthException -from .oauth import OAuth2Test, OAuth2PkcePlainTest, OAuth2PkceS256Test +from .oauth import OAuth2PkcePlainTest, OAuth2PkceS256Test, OAuth2Test class TwitterOAuth2Mixin: From bb0dd56ce4de5860dbaad0687061f70840700fcb Mon Sep 17 00:00:00 2001 From: Eshaan Bansal Date: Tue, 21 Nov 2023 18:52:29 +0530 Subject: [PATCH 11/13] chore: improvements, address review suggestions --- social_core/backends/bitbucket_datacenter.py | 8 +++++--- social_core/backends/oauth.py | 6 +++--- social_core/backends/twitter_oauth2.py | 2 +- social_core/tests/backends/oauth.py | 2 +- social_core/tests/backends/test_bitbucket_datacenter.py | 2 +- social_core/tests/backends/test_twitter_oauth2.py | 4 +--- 6 files changed, 12 insertions(+), 12 deletions(-) diff --git a/social_core/backends/bitbucket_datacenter.py b/social_core/backends/bitbucket_datacenter.py index 1569bb13e..d550b460d 100644 --- a/social_core/backends/bitbucket_datacenter.py +++ b/social_core/backends/bitbucket_datacenter.py @@ -43,8 +43,8 @@ class BitbucketDataCenterOAuth2(BaseOAuth2PKCE): ] PKCE_DEFAULT_CODE_CHALLENGE_METHOD = "s256" # can be "plain" or "s256" PKCE_DEFAULT_CODE_VERIFIER_LENGTH = 48 # must be b/w 43-127 chars - USE_PKCE = True - DEFAULT_AVATAR_SIZE = 48 + DEFAULT_USE_PKCE = True + DEFAULT_USER_AVATAR_SIZE = 48 @property def server_base_oauth2_api_url(self) -> str: @@ -100,6 +100,8 @@ def user_data(self, access_token, *args, **kwargs) -> dict: url=f"{self.server_base_rest_api_url}/users/{username}", headers=headers, params={ - "avatarSize": self.DEFAULT_AVATAR_SIZE # to force `avatarUrl` in response + "avatarSize": self.setting( + "USER_AVATAR_SIZE", default=self.DEFAULT_USER_AVATAR_SIZE + ) # to force `avatarUrl` in response }, ) diff --git a/social_core/backends/oauth.py b/social_core/backends/oauth.py index 86f0afe8e..2336f6ea7 100644 --- a/social_core/backends/oauth.py +++ b/social_core/backends/oauth.py @@ -476,7 +476,7 @@ class BaseOAuth2PKCE(BaseOAuth2): PKCE_DEFAULT_CODE_CHALLENGE_METHOD = "s256" PKCE_DEFAULT_CODE_VERIFIER_LENGTH = 32 - USE_PKCE = True + DEFAULT_USE_PKCE = True def create_code_verifier(self): name = self.name + "_code_verifier" @@ -507,7 +507,7 @@ def generate_code_challenge(self, code_verifier, challenge_method): def auth_params(self, state=None): params = super().auth_params(state=state) - if self.USE_PKCE: + if self.setting("USE_PKCE", default=self.DEFAULT_USE_PKCE): code_challenge_method = self.setting( "PKCE_CODE_CHALLENGE_METHOD", default=self.PKCE_DEFAULT_CODE_CHALLENGE_METHOD, @@ -523,7 +523,7 @@ def auth_params(self, state=None): def auth_complete_params(self, state=None): params = super().auth_complete_params(state=state) - if self.USE_PKCE: + if self.setting("USE_PKCE", default=self.DEFAULT_USE_PKCE): code_verifier = self.get_code_verifier() params["code_verifier"] = code_verifier diff --git a/social_core/backends/twitter_oauth2.py b/social_core/backends/twitter_oauth2.py index c56830dda..ead85b473 100644 --- a/social_core/backends/twitter_oauth2.py +++ b/social_core/backends/twitter_oauth2.py @@ -37,7 +37,7 @@ class TwitterOAuth2(BaseOAuth2PKCE): ] PKCE_DEFAULT_CODE_CHALLENGE_METHOD = "s256" PKCE_DEFAULT_CODE_VERIFIER_LENGTH = 32 - USE_PKCE = True + DEFAULT_USE_PKCE = True def get_user_details(self, response): """Return user details from Twitter account""" diff --git a/social_core/tests/backends/oauth.py b/social_core/tests/backends/oauth.py index f8bb483cc..c00e014f7 100644 --- a/social_core/tests/backends/oauth.py +++ b/social_core/tests/backends/oauth.py @@ -127,7 +127,7 @@ class OAuth2PkcePlainTest(OAuth2Test): def extra_settings(self): settings = super().extra_settings() settings.update( - {"SOCIAL_AUTH_" + self.name + "_PKCE_CODE_CHALLENGE_METHOD": "plain"} + {f"SOCIAL_AUTH_{self.name}_PKCE_CODE_CHALLENGE_METHOD": "plain"} ) return settings diff --git a/social_core/tests/backends/test_bitbucket_datacenter.py b/social_core/tests/backends/test_bitbucket_datacenter.py index aed0f07e7..8a3d3f9c5 100644 --- a/social_core/tests/backends/test_bitbucket_datacenter.py +++ b/social_core/tests/backends/test_bitbucket_datacenter.py @@ -62,7 +62,7 @@ class BitbucketDataCenterOAuth2Mixin: def extra_settings(self): settings = super().extra_settings() settings.update( - {"SOCIAL_AUTH_" + self.name + "_URL": "https://bachmanity.atlassian.net"} + {f"SOCIAL_AUTH_{self.name}_URL": "https://bachmanity.atlassian.net"} ) return settings diff --git a/social_core/tests/backends/test_twitter_oauth2.py b/social_core/tests/backends/test_twitter_oauth2.py index f54878958..5ed76abfd 100644 --- a/social_core/tests/backends/test_twitter_oauth2.py +++ b/social_core/tests/backends/test_twitter_oauth2.py @@ -184,9 +184,7 @@ class TwitterOAuth2TestInvalidCodeChallengeMethod( def test_login__error(self): self.strategy.set_settings( { - "SOCIAL_AUTH_" - + self.name - + "_PKCE_CODE_CHALLENGE_METHOD": "invalidmethodname" + f"SOCIAL_AUTH_{self.name}_PKCE_CODE_CHALLENGE_METHOD": "invalidmethodname", } ) From a0fc1d3a6cf3b713d1a41d5f6465494c4761c5ea Mon Sep 17 00:00:00 2001 From: Eshaan Bansal Date: Tue, 21 Nov 2023 18:53:41 +0530 Subject: [PATCH 12/13] fix: docs URL --- social_core/backends/bitbucket_datacenter.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/social_core/backends/bitbucket_datacenter.py b/social_core/backends/bitbucket_datacenter.py index d550b460d..419761486 100644 --- a/social_core/backends/bitbucket_datacenter.py +++ b/social_core/backends/bitbucket_datacenter.py @@ -1,6 +1,6 @@ """ Bitbucket Data Center OAuth2 backend, docs at: - https://python-social-auth.readthedocs.io/en/latest/backends/bitbucket-datacenter-oauth2.html + https://python-social-auth.readthedocs.io/en/latest/backends/bitbucket_datacenter_oauth2.html https://confluence.atlassian.com/bitbucketserver/bitbucket-oauth-2-0-provider-api-1108483661.html """ From 9799025203d91fa68155e93c71a2821ef31db2bb Mon Sep 17 00:00:00 2001 From: Eshaan Bansal Date: Wed, 22 Nov 2023 13:52:08 +0530 Subject: [PATCH 13/13] Apply suggestions from code review Co-authored-by: Johan Castiblanco <51926076+johanv26@users.noreply.github.com> --- social_core/backends/oauth.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/social_core/backends/oauth.py b/social_core/backends/oauth.py index 2336f6ea7..57e146a59 100644 --- a/social_core/backends/oauth.py +++ b/social_core/backends/oauth.py @@ -479,7 +479,7 @@ class BaseOAuth2PKCE(BaseOAuth2): DEFAULT_USE_PKCE = True def create_code_verifier(self): - name = self.name + "_code_verifier" + name = f"{self.name}_code_verifier" code_verifier_len = self.setting( "PKCE_CODE_VERIFIER_LENGTH", default=self.PKCE_DEFAULT_CODE_VERIFIER_LENGTH ) @@ -488,7 +488,7 @@ def create_code_verifier(self): return code_verifier def get_code_verifier(self): - name = self.name + "_code_verifier" + name = f"{self.name}_code_verifier" code_verifier = self.strategy.session_get(name) return code_verifier @@ -499,10 +499,9 @@ def generate_code_challenge(self, code_verifier, challenge_method): encoded = base64.urlsafe_b64encode(hashed) code_challenge = encoded.decode().replace("=", "") # remove padding return code_challenge - elif method == "plain": + if method == "plain": return code_verifier - else: - raise AuthException("Unsupported code challenge method.") + raise AuthException("Unsupported code challenge method.") def auth_params(self, state=None): params = super().auth_params(state=state)