Add jwt mgmt capabilities

Author: guyp-descope

descope/management/common.py

@@ -79,6 +79,9 @@ class MgmtV1:
     # jwt
     update_jwt_path = "/v1/mgmt/jwt/update"
     impersonate_path = "/v1/mgmt/impersonate"
+    mgmt_sign_in = "/v1/mgmt/auth/signin"
+    mgmt_sign_up = "/v1/mgmt/auth/signup"
+    mgmt_sign_up_or_in = "/v1/mgmt/auth/signup-in"
 
     # permission
     permission_create_path = "/v1/mgmt/permission/create"
@@ -146,6 +149,71 @@ class MgmtV1:
     project_list_projects = "/v1/mgmt/projects/list"
 
 
+class MgmtSignUpOptions:
+    def __init__(
+        self,
+        custom_claims: Optional[dict] = None,
+    ):
+        self.custom_claims = custom_claims
+
+
+class MgmtLoginOptions:
+    def __init__(
+        self,
+        stepup: bool = False,
+        mfa: bool = False,
+        revoke_other_sessions: Optional[bool] = None,
+        custom_claims: Optional[dict] = None,
+        jwt: Optional[str] = None,
+    ):
+        self.stepup = stepup
+        self.custom_claims = custom_claims
+        self.mfa = mfa
+        self.revoke_other_sessions = revoke_other_sessions
+        self.jwt = jwt
+
+
+def is_jwt_required(lgo: MgmtLoginOptions) -> bool:
+    return lgo is not None and (lgo.stepup or lgo.mfa)
+
+
+class MgmtUserRequest:
+    def __init__(
+        self,
+        name: Optional[str] = None,
+        given_name: Optional[str] = None,
+        middle_name: Optional[str] = None,
+        family_name: Optional[str] = None,
+        phone: Optional[str] = None,
+        email: Optional[str] = None,
+        email_verified: Optional[bool] = None,
+        phone_verified: Optional[bool] = None,
+        sso_app_id: Optional[str] = None,
+    ):
+        self.name = name
+        self.given_name = given_name
+        self.middle_name = middle_name
+        self.family_name = family_name
+        self.phone = phone
+        self.email = email
+        self.email_verified = email_verified
+        self.phone_verified = phone_verified
+        self.sso_app_id = sso_app_id
+
+    def to_dict(self) -> dict:
+        return {
+            "name": self.name,
+            "givenName": self.given_name,
+            "middleName": self.middle_name,
+            "familyName": self.family_name,
+            "phone": self.phone,
+            "email": self.email,
+            "emailVerified": self.email_verified,
+            "phoneVerified": self.phone_verified,
+            "ssoAppId": self.sso_app_id,
+        }
+
+
 class AssociatedTenant:
     """
     Represents a tenant association for a User or Access Key. The tenant_id is required to denote

descope/management/jwt.py

@@ -2,7 +2,13 @@
 
 from descope._auth_base import AuthBase
 from descope.exceptions import ERROR_TYPE_INVALID_ARGUMENT, AuthException
-from descope.management.common import MgmtV1
+from descope.management.common import (
+    MgmtLoginOptions,
+    MgmtSignUpOptions,
+    MgmtUserRequest,
+    MgmtV1,
+    is_jwt_required,
+)
 
 
 class JWT(AuthBase):
@@ -78,3 +84,91 @@ def impersonate(
             pswd=self._auth.management_key,
         )
         return response.json().get("jwt", "")
+
+    def sign_in(
+        self, login_id: str, login_options: Optional[MgmtLoginOptions] = None
+    ) -> dict:
+        """ """
+        if not login_id:
+            raise AuthException(
+                400, ERROR_TYPE_INVALID_ARGUMENT, "login_id cannot be empty"
+            )
+
+        if login_options is None:
+            login_options = MgmtLoginOptions()
+
+        if is_jwt_required(login_options) and not login_options.jwt:
+            raise AuthException(400, ERROR_TYPE_INVALID_ARGUMENT, "JWT is required")
+
+        response = self._auth.do_post(
+            MgmtV1.mgmt_sign_in,
+            {
+                "loginId": login_id,
+                "stepup": login_options.stepup,
+                "mfa": login_options.mfa,
+                "revokeOtherSessions": login_options.revoke_other_sessions,
+                "customClaims": login_options.custom_claims,
+                "jwt": login_options.jwt,
+            },
+            pswd=self._auth.management_key,
+        )
+        resp = response.json()
+        jwt_response = self._auth.generate_jwt_response(resp, None, None)
+        return jwt_response
+
+    def sign_up(
+        self,
+        login_id: str,
+        user: Optional[MgmtUserRequest] = None,
+        signup_options: Optional[MgmtSignUpOptions] = None,
+    ) -> dict:
+        """ """
+        return self._sign_up_internal(
+            login_id, MgmtV1.mgmt_sign_up, user, signup_options
+        )
+
+    def sign_up_or_in(
+        self,
+        login_id: str,
+        user: Optional[MgmtUserRequest] = None,
+        signup_options: Optional[MgmtSignUpOptions] = None,
+    ) -> dict:
+        """ """
+        return self._sign_up_internal(
+            login_id, MgmtV1.mgmt_sign_up_or_in, user, signup_options
+        )
+
+    def _sign_up_internal(
+        self,
+        login_id: str,
+        endpoint: str,
+        user: Optional[MgmtUserRequest] = None,
+        signup_options: Optional[MgmtSignUpOptions] = None,
+    ) -> dict:
+        """ """
+        if user is None:
+            user = MgmtUserRequest()
+
+        if not login_id:
+            raise AuthException(
+                400, ERROR_TYPE_INVALID_ARGUMENT, "login_id cannot be empty"
+            )
+
+        if signup_options is None:
+            signup_options = MgmtSignUpOptions()
+
+        response = self._auth.do_post(
+            endpoint,
+            {
+                "loginId": login_id,
+                "user": user.to_dict(),
+                "emailVerified": user.email_verified,
+                "phoneVerified": user.phone_verified,
+                "ssoAppId": user.sso_app_id,
+                "customClaims": signup_options.custom_claims,
+            },
+            pswd=self._auth.management_key,
+        )
+        resp = response.json()
+        jwt_response = self._auth.generate_jwt_response(resp, None, None)
+        return jwt_response

samples/management/jwt_mgmt.py

@@ -0,0 +1,39 @@
+import logging
+
+from descope import AuthException, DescopeClient
+
+logging.basicConfig(level=logging.INFO)
+
+
+def main():
+    project_id = ""
+    management_key = ""
+
+    try:
+        descope_client = DescopeClient(
+            project_id=project_id, management_key=management_key
+        )
+        user_login_id = "[email protected]"
+
+        try:
+            logging.info("Going to create a new user by sign up or in method")
+            resp = descope_client.mgmt.jwt.sign_up_or_in(user_login_id)
+            logging.info(f"Response: {resp}")
+
+        except AuthException as e:
+            logging.info(f"User SignUpOrIn failed {e}")
+
+        try:
+            logging.info("Searching for created user")
+            resp = descope_client.mgmt.jwt.sign_in(user_login_id)
+            logging.info(f"Response: {resp}")
+
+        except AuthException as e:
+            logging.info(f"User SignIn failed {e}")
+
+    except AuthException:
+        raise
+
+
+if __name__ == "__main__":
+    main()

tests/management/test_jwt.py

@@ -4,7 +4,7 @@
 
 from descope import AuthException, DescopeClient
 from descope.common import DEFAULT_TIMEOUT_SECONDS
-from descope.management.common import MgmtV1
+from descope.management.common import MgmtLoginOptions, MgmtV1
 
 from .. import common
 
@@ -119,3 +119,147 @@ def test_impersonate(self):
                 params=None,
                 timeout=DEFAULT_TIMEOUT_SECONDS,
             )
+
+    def test_sign_in(self):
+        client = DescopeClient(
+            self.dummy_project_id,
+            self.public_key_dict,
+            False,
+            self.dummy_management_key,
+        )
+
+        # Test failed flows
+        self.assertRaises(AuthException, client.mgmt.jwt.sign_in, "")
+
+        self.assertRaises(
+            AuthException,
+            client.mgmt.jwt.sign_in,
+            "loginId",
+            MgmtLoginOptions(mfa=True),
+        )
+
+        # Test success flow
+        with patch("requests.post") as mock_post:
+            network_resp = mock.Mock()
+            network_resp.ok = True
+            network_resp.json.return_value = json.loads("""{"jwt": "response"}""")
+            mock_post.return_value = network_resp
+            client.mgmt.jwt.sign_in("loginId")
+            expected_uri = f"{common.DEFAULT_BASE_URL}{MgmtV1.mgmt_sign_in}"
+            mock_post.assert_called_with(
+                expected_uri,
+                headers={
+                    **common.default_headers,
+                    "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
+                },
+                json={
+                    "loginId": "loginId",
+                    "stepup": False,
+                    "mfa": False,
+                    "revokeOtherSessions": None,
+                    "customClaims": None,
+                    "jwt": None,
+                },
+                allow_redirects=False,
+                verify=True,
+                params=None,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )
+
+    def test_sign_up(self):
+        client = DescopeClient(
+            self.dummy_project_id,
+            self.public_key_dict,
+            False,
+            self.dummy_management_key,
+        )
+
+        # Test failed flows
+        self.assertRaises(AuthException, client.mgmt.jwt.sign_up, "")
+
+        # Test success flow
+        with patch("requests.post") as mock_post:
+            network_resp = mock.Mock()
+            network_resp.ok = True
+            network_resp.json.return_value = json.loads("""{"jwt": "response"}""")
+            mock_post.return_value = network_resp
+            client.mgmt.jwt.sign_up("loginId")
+            expected_uri = f"{common.DEFAULT_BASE_URL}{MgmtV1.mgmt_sign_up}"
+            mock_post.assert_called_with(
+                expected_uri,
+                headers={
+                    **common.default_headers,
+                    "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
+                },
+                json={
+                    "loginId": "loginId",
+                    "user": {
+                        "name": None,
+                        "givenName": None,
+                        "middleName": None,
+                        "familyName": None,
+                        "phone": None,
+                        "email": None,
+                        "emailVerified": None,
+                        "phoneVerified": None,
+                        "ssoAppId": None,
+                    },
+                    "emailVerified": None,
+                    "phoneVerified": None,
+                    "ssoAppId": None,
+                    "customClaims": None,
+                },
+                allow_redirects=False,
+                verify=True,
+                params=None,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )
+
+    def test_sign_up_or_in(self):
+        client = DescopeClient(
+            self.dummy_project_id,
+            self.public_key_dict,
+            False,
+            self.dummy_management_key,
+        )
+
+        # Test failed flows
+        self.assertRaises(AuthException, client.mgmt.jwt.sign_up_or_in, "")
+
+        # Test success flow
+        with patch("requests.post") as mock_post:
+            network_resp = mock.Mock()
+            network_resp.ok = True
+            network_resp.json.return_value = json.loads("""{"jwt": "response"}""")
+            mock_post.return_value = network_resp
+            client.mgmt.jwt.sign_up_or_in("loginId")
+            expected_uri = f"{common.DEFAULT_BASE_URL}{MgmtV1.mgmt_sign_up_or_in}"
+            mock_post.assert_called_with(
+                expected_uri,
+                headers={
+                    **common.default_headers,
+                    "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
+                },
+                json={
+                    "loginId": "loginId",
+                    "user": {
+                        "name": None,
+                        "givenName": None,
+                        "middleName": None,
+                        "familyName": None,
+                        "phone": None,
+                        "email": None,
+                        "emailVerified": None,
+                        "phoneVerified": None,
+                        "ssoAppId": None,
+                    },
+                    "emailVerified": None,
+                    "phoneVerified": None,
+                    "ssoAppId": None,
+                    "customClaims": None,
+                },
+                allow_redirects=False,
+                verify=True,
+                params=None,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )