diff --git a/social_core/backends/echosign.py b/social_core/backends/echosign.py deleted file mode 100644 index ac061dc77..000000000 --- a/social_core/backends/echosign.py +++ /dev/null @@ -1,26 +0,0 @@ -from typing import Any - -from .oauth import BaseOAuth2 - - -class EchosignOAuth2(BaseOAuth2): - name = "echosign" - REDIRECT_STATE = False - REFRESH_TOKEN_METHOD = "POST" - REVOKE_TOKEN_METHOD = "POST" - AUTHORIZATION_URL = "https://secure.echosign.com/public/oauth" - ACCESS_TOKEN_URL = "https://secure.echosign.com/oauth/token" - REFRESH_TOKEN_URL = "https://secure.echosign.com/oauth/refresh" - REVOKE_TOKEN_URL = "https://secure.echosign.com/oauth/revoke" - - def get_user_details(self, response): - return response - - def get_user_id(self, details, response): - return details["userInfoList"][0]["userId"] - - def user_data(self, access_token: str, *args, **kwargs) -> dict[str, Any] | None: - return self.get_json( - "https://api.echosign.com/api/rest/v3/users", - headers={"Access-Token": access_token}, - ) diff --git a/social_core/backends/exacttarget.py b/social_core/backends/exacttarget.py deleted file mode 100644 index c2dc061d2..000000000 --- a/social_core/backends/exacttarget.py +++ /dev/null @@ -1,117 +0,0 @@ -""" -ExactTarget OAuth support. -Support Authentication from IMH using JWT token and pre-shared key. -Requires package pyjwt -""" - -from __future__ import annotations - -from datetime import datetime, timedelta, timezone -from typing import Any - -import jwt - -from social_core.exceptions import AuthCanceled, AuthFailed - -from .oauth import BaseOAuth2 - - -class ExactTargetOAuth2(BaseOAuth2): - name = "exacttarget" - - def get_user_details(self, response): - """Use the email address of the user, suffixed by _et""" - user = response.get("token", {}).get("request", {}).get("user", {}) - if "email" in user: - user["username"] = user["email"] - return user - - def get_user_id(self, details, response): - """ - Create a user ID from the ET user ID. Uses details rather than the - default response, as only the token is available in response. details - is much richer: - { - 'expiresIn': 1200, - 'username': 'example@example.com', - 'refreshToken': '1234567890abcdef', - 'internalOauthToken': 'jwttoken.......', - 'oauthToken': 'yetanothertoken', - 'id': 123456, - 'culture': 'en-US', - 'timezone': { - 'shortName': 'CST', - 'offset': -6.0, - 'dst': False, - 'longName': '(GMT-06:00) Central Time (No Daylight Saving)' - }, - 'email': 'example@example.com' - } - """ - return str(details.get("id")) - - def uses_redirect(self) -> bool: - return False - - def auth_url(self) -> str: - return "" - - def process_error(self, data) -> None: - if data.get("error"): - error = self.data.get("error_description") or self.data["error"] - raise AuthFailed(self, error) - - def do_auth(self, token, *args, **kwargs): - _key, secret = self.get_key_and_secret() - try: # Decode the token, using the Application Signature from settings - decoded = jwt.decode(token, secret, algorithms=["HS256"]) - except jwt.PyJWTError as error: - # Wrong signature, fail authentication - raise AuthCanceled(self) from error - kwargs.update({"response": {"token": decoded}, "backend": self}) - return self.strategy.authenticate(*args, **kwargs) - - def auth_complete(self, *args, **kwargs): - """Completes login process, must return user instance""" - token = self.data.get("jwt", {}) - if not token: - raise AuthFailed(self, "Authentication Failed") - return self.do_auth(token, *args, **kwargs) - - def extra_data( - self, - user, - uid: str, - response: dict[str, Any], - details: dict[str, Any], - pipeline_kwargs: dict[str, Any], - ) -> dict[str, Any]: - """Load extra details from the JWT token""" - data = { - "id": details.get("id"), - "email": details.get("email"), - # OAuth token, for use with legacy SOAP API calls: - # http://bit.ly/13pRHfo - "internalOauthToken": details.get("internalOauthToken"), - # Token for use with the Application ClientID for the FUEL API - "oauthToken": details.get("oauthToken"), - # If the token has expired, use the FUEL API to get a new token see - # http://bit.ly/10v1K5l and http://bit.ly/11IbI6F - set legacy=1 - "refreshToken": details.get("refreshToken"), - } - - # The expiresIn value determines how long the tokens are valid for. - # Take a bit off, then convert to an int timestamp - expiresSeconds = details.get("expiresIn", 0) - 30 - expires = datetime.now(timezone.utc) + timedelta(seconds=expiresSeconds) - data["expires"] = ( - expires - datetime(1970, 1, 1, tzinfo=timezone.utc) - ).total_seconds() - - if response.get("token"): - token = response["token"] - org = token.get("request", {}).get("organization") - if org: - data["stack"] = org.get("stackKey") - data["enterpriseId"] = org.get("enterpriseId") - return data diff --git a/social_core/backends/pocket.py b/social_core/backends/pocket.py deleted file mode 100644 index 820f787f6..000000000 --- a/social_core/backends/pocket.py +++ /dev/null @@ -1,57 +0,0 @@ -""" -Pocket OAuth2 backend, docs at: - https://python-social-auth.readthedocs.io/en/latest/backends/pocket.html -""" - -from __future__ import annotations - -from typing import Any - -from social_core.utils import handle_http_errors - -from .base import BaseAuth - - -class PocketAuth(BaseAuth): - name = "pocket" - AUTHORIZATION_URL = "https://getpocket.com/auth/authorize" - ACCESS_TOKEN_URL = "https://getpocket.com/v3/oauth/authorize" - REQUEST_TOKEN_URL = "https://getpocket.com/v3/oauth/request" - ID_KEY = "username" - - def get_json(self, url, *args, **kwargs): - headers = {"X-Accept": "application/json"} - kwargs.update({"method": "POST", "headers": headers}) - return super().get_json(url, *args, **kwargs) - - def get_user_details(self, response): - return {"username": response["username"]} - - def extra_data( - self, - user, - uid: str, - response: dict[str, Any], - details: dict[str, Any], - pipeline_kwargs: dict[str, Any], - ) -> dict[str, Any]: - return response - - def auth_url(self) -> str: - data = { - "consumer_key": self.setting("KEY"), - "redirect_uri": self.redirect_uri, - } - token = self.get_json(self.REQUEST_TOKEN_URL, data=data)["code"] - self.strategy.session_set("pocket_request_token", token) - return f"{self.AUTHORIZATION_URL}?request_token={token}&redirect_uri={self.redirect_uri}" - - @handle_http_errors - def auth_complete(self, *args, **kwargs): - data = { - "consumer_key": self.setting("KEY"), - "code": self.strategy.session_get("pocket_request_token"), - } - response = self.get_json(self.ACCESS_TOKEN_URL, data=data) - kwargs.update({"response": response, "backend": self}) - return self.strategy.authenticate(*args, **kwargs) diff --git a/social_core/backends/runkeeper.py b/social_core/backends/runkeeper.py deleted file mode 100644 index 3a7bf2ab4..000000000 --- a/social_core/backends/runkeeper.py +++ /dev/null @@ -1,52 +0,0 @@ -""" -RunKeeper OAuth2 backend, docs at: - https://python-social-auth.readthedocs.io/en/latest/backends/runkeeper.html -""" - -from typing import Any - -from .oauth import BaseOAuth2 - - -class RunKeeperOAuth2(BaseOAuth2): - """RunKeeper OAuth authentication backend""" - - name = "runkeeper" - AUTHORIZATION_URL = "https://runkeeper.com/apps/authorize" - ACCESS_TOKEN_URL = "https://runkeeper.com/apps/token" - EXTRA_DATA = [ - ("userID", "id"), - ] - - def get_user_id(self, details, response): - return response["userID"] - - def get_user_details(self, response): - """Parse username from profile link""" - username = None - profile_url = response.get("profile") - if len(profile_url): - profile_url_parts = profile_url.split("http://runkeeper.com/user/") - if len(profile_url_parts) > 1 and len(profile_url_parts[1]): - username = profile_url_parts[1] - fullname, first_name, last_name = self.get_user_names( - fullname=response.get("name") - ) - return { - "username": username, - "email": response.get("email") or "", - "fullname": fullname, - "first_name": first_name, - "last_name": last_name, - } - - def user_data(self, access_token: str, *args, **kwargs) -> dict[str, Any] | None: - # We need to use the /user endpoint to get the user id, the /profile - # endpoint contains name, user name, location, gender - user_data = self._user_data(access_token, "/user") - profile_data = self._user_data(access_token, "/profile") - return dict(user_data, **profile_data) - - def _user_data(self, access_token, path): - url = f"https://api.runkeeper.com{path}" - return self.get_json(url, params={"access_token": access_token}) diff --git a/social_core/tests/backends/test_exacttarget.py b/social_core/tests/backends/test_exacttarget.py deleted file mode 100644 index feafb4d36..000000000 --- a/social_core/tests/backends/test_exacttarget.py +++ /dev/null @@ -1,28 +0,0 @@ -from unittest.mock import patch - -import jwt - -from social_core.exceptions import AuthCanceled - -from .base import BaseBackendTest - - -class ExactTargetOAuth2Test(BaseBackendTest): - backend_path = "social_core.backends.exacttarget.ExactTargetOAuth2" - - def extra_settings(self) -> dict[str, str | list[str]]: - return { - "SOCIAL_AUTH_EXACTTARGET_KEY": "key", - "SOCIAL_AUTH_EXACTTARGET_SECRET": "secret", - } - - def test_jwt_error_is_wrapped(self) -> None: - error = jwt.ExpiredSignatureError("expired") - - with ( - patch("social_core.backends.exacttarget.jwt.decode", side_effect=error), - self.assertRaises(AuthCanceled) as context, - ): - self.backend.do_auth("token") - - self.assertIs(context.exception.__cause__, error)