Support OAuth authentication method (#5)

* Support OAuth authentication method

Author: guyp-descope

descope/auth.py

@@ -20,6 +20,7 @@
     SESSION_COOKIE_NAME,
     DeliveryMethod,
     EndpointsV1,
+    OAuthProviders,
     User,
 )
 from descope.exceptions import AuthException
@@ -614,6 +615,14 @@ def validate_session_request(
     def logout(
         self, signed_token: str, signed_refresh_token: str
     ) -> requests.cookies.RequestsCookieJar:
+
+        if signed_token is None or signed_refresh_token is None:
+            raise AuthException(
+                401,
+                "token validation failure",
+                f"signed token {signed_token} or/and signed refresh token {signed_refresh_token} are empty",
+            )
+
         uri = AuthClient._compose_logout_url()
         cookies = {
             SESSION_COOKIE_NAME: signed_token,
@@ -642,3 +651,38 @@ def _get_default_headers(self):
         bytes = f"{self.project_id}:".encode("ascii")
         headers["Authorization"] = f"Basic {base64.b64encode(bytes).decode('ascii')}"
         return headers
+
+    @staticmethod
+    def _verify_oauth_provider(oauth_provider: str) -> str:
+        if oauth_provider == "" or oauth_provider is None:
+            return False
+
+        if oauth_provider in OAuthProviders:
+            return True
+        else:
+            return False
+
+    def oauth_start(self, provider: str) -> str:
+        """ """
+        if not self._verify_oauth_provider(provider):
+            raise AuthException(
+                500,
+                "Unknown OAuth provider",
+                f"Unknown OAuth provider: {provider}",
+            )
+
+        uri = f"{DEFAULT_BASE_URI}{EndpointsV1.oauthStart}"
+        response = requests.get(
+            uri,
+            headers=self._get_default_headers(),
+            params={"provider": provider},
+            allow_redirects=False,
+        )
+
+        if not response.ok:
+            raise AuthException(
+                response.status_code, "OAuth send request failure", response.text
+            )
+
+        redirect_url = response.headers.get("Location", "")
+        return redirect_url

descope/common.py

@@ -8,6 +8,8 @@
 SESSION_COOKIE_NAME = "DS"
 REFRESH_SESSION_COOKIE_NAME = "DSR"
 
+REDIRECT_LOCATION_COOKIE_NAME = "Location"
+
 
 class EndpointsV1:
     signInAuthOTPPath = "/v1/auth/signin/otp"
@@ -16,6 +18,7 @@ class EndpointsV1:
     signInAuthMagicLinkPath = "/v1/auth/signin/magiclink"
     signUpAuthMagicLinkPath = "/v1/auth/signup/magiclink"
     verifyMagicLinkAuthPath = "/v1/auth/magiclink/verify"
+    oauthStart = "/v1/oauth/authorize"
     publicKeyPath = "/v1/keys"
     refreshTokenPath = "/v1/auth/refresh"
     logoutPath = "/v1/auth/logoutall"
@@ -27,6 +30,9 @@ class DeliveryMethod(Enum):
     EMAIL = 3
 
 
+OAuthProviders = ["facebook", "github", "google", "microsoft", "gitlab", "apple"]
+
+
 class User:
     def __init__(self, username: str, name: str, phone: str, email: str):
         self.username = username

samples/decorators/flask_decorators.py

@@ -2,7 +2,7 @@
 import sys
 from functools import wraps
 
-from flask import Response, _request_ctx_stack, request
+from flask import Response, _request_ctx_stack, redirect, request
 
 dir_name = os.path.dirname(__file__)
 sys.path.insert(0, os.path.join(dir_name, "../"))
@@ -329,3 +329,29 @@ def decorated(*args, **kwargs):
         return decorated
 
     return decorator
+
+
+def descope_oauth(auth_client):
+    """
+    OAuth login
+    """
+
+    def decorator(f):
+        @wraps(f)
+        def decorated(*args, **kwargs):
+            try:
+                args = request.args
+                provider = args.get("provider")
+                redirect_url = auth_client.oauth_start(provider)
+            except AuthException as e:
+                return Response(f"OAuth failed {e}", e.status_code)
+
+            # Execute the original API
+            # (ignore return value as anyway we redirect)
+            f(*args, **kwargs)
+
+            return redirect(redirect_url, 302)
+
+        return decorated
+
+    return decorator

samples/oauth_web_sample_app.py

@@ -0,0 +1,65 @@
+import os
+import sys
+
+from flask import Flask, jsonify
+
+dir_name = os.path.dirname(__file__)
+sys.path.insert(0, os.path.join(dir_name, "../"))
+from decorators.flask_decorators import (  # noqa: E402;
+    descope_logout,
+    descope_oauth,
+    descope_validate_auth,
+)
+
+from descope import AuthClient  # noqa: E402
+
+APP = Flask(__name__)
+
+PROJECT_ID = ""
+
+# init the AuthClient
+auth_client = AuthClient(PROJECT_ID)
+
+
+class Error(Exception):
+    def __init__(self, error, status_code):
+        self.error = error
+        self.status_code = status_code
+
+
+@APP.errorhandler(Error)
+def handle_auth_error(ex):
+    response = jsonify(ex.error)
+    response.status_code = ex.status_code
+    return response
+
+
+# This needs authentication
+@APP.route("/api/private")
+@descope_validate_auth(auth_client)
+def private():
+    response = "This is a private API and you must be authenticated to see this"
+    return jsonify(message=response)
+
+
+@APP.route("/api/logout")
+@descope_logout(auth_client)
+def logout():
+    response = "Logged out"
+    return jsonify(message=response)
+
+
+@APP.route("/api/oauth", methods=["GET"])
+@descope_oauth(auth_client)
+def oauth(*args, **kwargs):
+    pass
+
+
+# This doesn't need authentication
+@APP.route("/")
+def home():
+    return "OK"
+
+
+if __name__ == "__main__":
+    APP.run(host="127.0.0.1", port=9000)

tests/test_auth.py

@@ -5,7 +5,7 @@
 from unittest.mock import patch
 
 from descope import SESSION_COOKIE_NAME, AuthClient, AuthException, DeliveryMethod, User
-from descope.common import REFRESH_SESSION_COOKIE_NAME
+from descope.common import DEFAULT_BASE_URI, REFRESH_SESSION_COOKIE_NAME, EndpointsV1
 
 
 class TestAuthClient(unittest.TestCase):
@@ -172,6 +172,56 @@ class AAA(Enum):
             False,
         )
 
+    def test_verify_oauth_providers(self):
+        self.assertEqual(
+            AuthClient._verify_oauth_provider(""),
+            False,
+        )
+
+        self.assertEqual(
+            AuthClient._verify_oauth_provider(None),
+            False,
+        )
+
+        self.assertEqual(
+            AuthClient._verify_oauth_provider("unknown provider"),
+            False,
+        )
+
+        self.assertEqual(
+            AuthClient._verify_oauth_provider("google"),
+            True,
+        )
+
+    def test_oauth_start(self):
+        client = AuthClient(self.dummy_project_id, self.public_key_dict)
+
+        # Test failed flows
+        self.assertRaises(AuthException, client.oauth_start, "")
+
+        with patch("requests.get") as mock_get:
+            mock_get.return_value.ok = False
+            self.assertRaises(AuthException, client.oauth_start, "google")
+
+        # Test success flow
+        with patch("requests.get") as mock_get:
+            mock_get.return_value.ok = True
+            self.assertIsNotNone(client.oauth_start("google"))
+
+        with patch("requests.get") as mock_get:
+            mock_get.return_value.ok = True
+            client.oauth_start("facebook")
+            expected_uri = f"{DEFAULT_BASE_URI}{EndpointsV1.oauthStart}"
+            mock_get.assert_called_with(
+                expected_uri,
+                headers={
+                    "Content-Type": "application/json",
+                    "Authorization": "Basic ZHVtbXk6",
+                },
+                params={"provider": "facebook"},
+                allow_redirects=False,
+            )
+
     def test_get_identifier_name_by_method(self):
         self.assertEqual(
             AuthClient._get_identifier_name_by_method(DeliveryMethod.EMAIL), "email"
@@ -277,6 +327,8 @@ def test_logout(self):
         dummy_valid_jwt_token = ""
         client = AuthClient(self.dummy_project_id, self.public_key_dict)
 
+        self.assertRaises(AuthException, client.logout, None, None)
+
         # Test failed flow
         with patch("requests.get") as mock_get:
             mock_get.return_value.ok = False