Merge branch 'main' of github.com:descope/python-sdk into return-user-response

Author: dorsha

.vscode/settings.json

@@ -4,5 +4,10 @@
   "flake8.importStrategy": "fromEnvironment",
   "mypy-type-checker.importStrategy": "fromEnvironment",
   "isort.importStrategy": "fromEnvironment",
-  "black-formatter.importStrategy": "fromEnvironment"
+  "black-formatter.importStrategy": "fromEnvironment",
+  "workbench.colorCustomizations": {
+    "activityBar.background": "#4D1C3B",
+    "titleBar.activeBackground": "#6B2752",
+    "titleBar.activeForeground": "#FDF8FB"
+  }
 }

README.md

@@ -282,6 +282,14 @@ refresh_token = jwt_response[REFRESH_SESSION_TOKEN_NAME].get("jwt")
 
 The session and refresh JWTs should be returned to the caller, and passed with every request in the session. Read more on [session validation](#session-validation)
 
+#### Deleting the TOTP Seed
+
+Pass the loginId to the function to remove the user's TOTP seed.
+
+```python
+response = descope_client.mgmt.user.remove_totp_seed(login_id=login_id)
+```
+
 ### Passwords
 
 The user can also authenticate with a password, though it's recommended to

descope/auth.py

@@ -105,18 +105,34 @@ def __init__(
                 self.public_keys = {kid: (pub_key, alg)}
 
     def _raise_rate_limit_exception(self, response):
-        resp = response.json()
-        raise RateLimitException(
-            resp.get("errorCode", HTTPStatus.TOO_MANY_REQUESTS),
-            ERROR_TYPE_API_RATE_LIMIT,
-            resp.get("errorDescription", ""),
-            resp.get("errorMessage", ""),
-            rate_limit_parameters={
-                API_RATE_LIMIT_RETRY_AFTER_HEADER: int(
-                    response.headers.get(API_RATE_LIMIT_RETRY_AFTER_HEADER, 0)
-                )
-            },
-        )
+        try:
+            resp = response.json()
+            raise RateLimitException(
+                resp.get("errorCode", HTTPStatus.TOO_MANY_REQUESTS),
+                ERROR_TYPE_API_RATE_LIMIT,
+                resp.get("errorDescription", ""),
+                resp.get("errorMessage", ""),
+                rate_limit_parameters={
+                    API_RATE_LIMIT_RETRY_AFTER_HEADER: self._parse_retry_after(
+                        response.headers
+                    )
+                },
+            )
+        except RateLimitException:
+            raise
+        except Exception as e:
+            raise RateLimitException(
+                status_code=HTTPStatus.TOO_MANY_REQUESTS,
+                error_type=ERROR_TYPE_API_RATE_LIMIT,
+                error_message=ERROR_TYPE_API_RATE_LIMIT,
+                error_description=ERROR_TYPE_API_RATE_LIMIT,
+            )
+
+    def _parse_retry_after(self, headers):
+        try:
+            return int(headers.get(API_RATE_LIMIT_RETRY_AFTER_HEADER, 0))
+        except (ValueError, TypeError):
+            return 0
 
     def do_get(
         self,

descope/management/common.py

@@ -80,9 +80,10 @@ 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"
+    mgmt_sign_in_path = "/v1/mgmt/auth/signin"
+    mgmt_sign_up_path = "/v1/mgmt/auth/signup"
+    mgmt_sign_up_or_in_path = "/v1/mgmt/auth/signup-in"
+    anonymous_path = "/v1/mgmt/auth/anonymous"
 
     # permission
     permission_create_path = "/v1/mgmt/permission/create"

descope/management/jwt.py

@@ -13,7 +13,7 @@
 
 class JWT(AuthBase):
     def update_jwt(
-        self, jwt: str, custom_claims: dict, refresh_duration: Optional[int]
+        self, jwt: str, custom_claims: dict, refresh_duration: int = 0
     ) -> str:
         """
         Given a valid JWT, update it with custom claims, and update its authz claims as well
@@ -108,7 +108,7 @@ def sign_in(
             raise AuthException(400, ERROR_TYPE_INVALID_ARGUMENT, "JWT is required")
 
         response = self._auth.do_post(
-            MgmtV1.mgmt_sign_in,
+            MgmtV1.mgmt_sign_in_path,
             {
                 "loginId": login_id,
                 "stepup": login_options.stepup,
@@ -139,7 +139,7 @@ def sign_up(
         """
 
         return self._sign_up_internal(
-            login_id, MgmtV1.mgmt_sign_up, user, signup_options
+            login_id, MgmtV1.mgmt_sign_up_path, user, signup_options
         )
 
     def sign_up_or_in(
@@ -157,7 +157,7 @@ def sign_up_or_in(
         signup_options (MgmtSignUpOptions): signup options.
         """
         return self._sign_up_internal(
-            login_id, MgmtV1.mgmt_sign_up_or_in, user, signup_options
+            login_id, MgmtV1.mgmt_sign_up_or_in_path, user, signup_options
         )
 
     def _sign_up_internal(
@@ -193,3 +193,30 @@ def _sign_up_internal(
         resp = response.json()
         jwt_response = self._auth.generate_jwt_response(resp, None, None)
         return jwt_response
+
+    def anonymous(
+        self,
+        custom_claims: Optional[dict] = None,
+        tenant_id: Optional[str] = None,
+    ) -> dict:
+        """
+        Generate a JWT for an anonymous user.
+
+        Args:
+        custom_claims dict: Custom claims to add to JWT
+        tenant_id (str): tenant id to set on DCT claim.
+        """
+
+        response = self._auth.do_post(
+            MgmtV1.anonymous_path,
+            {
+                "customClaims": custom_claims,
+                "selectedTenant": tenant_id,
+            },
+            pswd=self._auth.management_key,
+        )
+        resp = response.json()
+        jwt_response = self._auth.generate_jwt_response(resp, None, None)
+        del jwt_response["firstSeen"]
+        del jwt_response["user"]
+        return jwt_response

descope/management/user.py

@@ -611,6 +611,7 @@ def search_all(
         to_created_time: Optional[int] = None,
         from_modified_time: Optional[int] = None,
         to_modified_time: Optional[int] = None,
+        user_ids: Optional[List[str]] = None,
     ) -> dict:
         """
         Search all users.
@@ -634,6 +635,7 @@ def search_all(
         to_created_time (int): Optional int, only include users who were created on or before this time (in Unix epoch milliseconds)
         from_modified_time (int): Optional int, only include users whose last modification/update occurred on or after this time (in Unix epoch milliseconds)
         to_modified_time (int): Optional int, only include users whose last modification/update occurred on or before this time (in Unix epoch milliseconds)
+        user_ids (List[str]): Optional list of user IDs to filter by
 
         Return value (dict):
         Return dict in the format
@@ -681,6 +683,9 @@ def search_all(
         if login_ids is not None:
             body["loginIds"] = login_ids
 
+        if user_ids is not None:
+            body["userIds"] = user_ids
+
         if text is not None:
             body["text"] = text
 

poetry.lock

@@ -234,9 +234,9 @@ importlib-metadata==8.5.0 ; python_full_version >= "3.8.1" and python_version <
 itsdangerous==2.2.0 ; python_full_version >= "3.8.1" and python_version < "4.0" \
     --hash=sha256:c6242fc49e35958c8b15141343aa660db5fc54d4f13a1db01a3f5891b98700ef \
     --hash=sha256:e0050c0b7da1eea53ffaf149c0cfbb5c6e2e2b69c4bef22c81fa6eb73e5f6173
-Jinja2==3.1.5; python_full_version >= "3.8.1" and python_version < "4.0" \
-    --hash=sha256:8fefff8dc3034e27bb80d67c671eb8a9bc424c0ef4c0826edbff304cceff43bb \
-    --hash=sha256:aba0f4dc9ed8013c424088f68a5c226f7d6097ed89b246d7749c2ec4175c6adb
+Jinja2==3.1.6; python_full_version >= "3.8.1" and python_version < "4.0" \
+    --hash=sha256:0137fb05990d35f1275a587e9aee6d56da821fc83491a0fb838183be43f66d6d \
+    --hash=sha256:85ece4451f492d0c13c5dd7c13a64681a86afae63a5f347908daf103ce6d2f67
 liccheck==0.9.2 ; python_full_version >= "3.8.1" and python_version < "4.0" \
     --hash=sha256:15cbedd042515945fe9d58b62e0a5af2f2a7795def216f163bb35b3016a16637 \
     --hash=sha256:bdc2190f8e95af3c8f9c19edb784ba7d41ecb2bf9189422eae6112bf84c08cd5

requirements.txt

@@ -69,6 +69,26 @@ def test_update_jwt(self):
                 timeout=DEFAULT_TIMEOUT_SECONDS,
             )
 
+            resp = client.mgmt.jwt.update_jwt("test", {"k1": "v1"})
+            self.assertEqual(resp, "response")
+            expected_uri = f"{common.DEFAULT_BASE_URL}{MgmtV1.update_jwt_path}"
+            mock_post.assert_called_with(
+                expected_uri,
+                headers={
+                    **common.default_headers,
+                    "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
+                },
+                json={
+                    "jwt": "test",
+                    "customClaims": {"k1": "v1"},
+                    "refreshDuration": 0,
+                },
+                allow_redirects=False,
+                verify=True,
+                params=None,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )
+
     def test_impersonate(self):
         client = DescopeClient(
             self.dummy_project_id,
@@ -145,7 +165,7 @@ def test_sign_in(self):
             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}"
+            expected_uri = f"{common.DEFAULT_BASE_URL}{MgmtV1.mgmt_sign_in_path}"
             mock_post.assert_called_with(
                 expected_uri,
                 headers={
@@ -184,7 +204,7 @@ def test_sign_up(self):
             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}"
+            expected_uri = f"{common.DEFAULT_BASE_URL}{MgmtV1.mgmt_sign_up_path}"
             mock_post.assert_called_with(
                 expected_uri,
                 headers={
@@ -233,7 +253,7 @@ def test_sign_up_or_in(self):
             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}"
+            expected_uri = f"{common.DEFAULT_BASE_URL}{MgmtV1.mgmt_sign_up_or_in_path}"
             mock_post.assert_called_with(
                 expected_uri,
                 headers={
@@ -263,3 +283,32 @@ def test_sign_up_or_in(self):
                 params=None,
                 timeout=DEFAULT_TIMEOUT_SECONDS,
             )
+
+    def test_anonymous(self):
+        client = DescopeClient(
+            self.dummy_project_id,
+            self.public_key_dict,
+            False,
+            self.dummy_management_key,
+        )
+
+        # 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.anonymous({"k1": "v1"}, "id")
+            expected_uri = f"{common.DEFAULT_BASE_URL}{MgmtV1.anonymous_path}"
+            mock_post.assert_called_with(
+                expected_uri,
+                headers={
+                    **common.default_headers,
+                    "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
+                },
+                json={"customClaims": {"k1": "v1"}, "selectedTenant": "id"},
+                allow_redirects=False,
+                verify=True,
+                params=None,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )