Merge pull request #454 from TheHive-Project/436-review-user-endpoints

#436 - Review user endpoints

Author: Kamforka

docs/release-notes.md

@@ -30,7 +30,7 @@
 * [#425](https://github.com/TheHive-Project/TheHive4py/issues/425) - Use TheHive v5.5.2 in integration tests by [@Kamforka](https://github.com/Kamforka) in [#433](https://github.com/TheHive-Project/TheHive4py/pull/433)
 * [#438](https://github.com/TheHive-Project/TheHive4py/issues/438) - Add docstrings to the `task_log` endpoints by [@Kamforka](https://github.com/Kamforka) in [#451](https://github.com/TheHive-Project/TheHive4py/pull/451)
 
-## New Contributors
+### New Contributors
 * [@3lina](https://github.com/3lina) made their first contribution in [#430](https://github.com/TheHive-Project/TheHive4py/pull/430)
 
 **Full Changelog**: [2.0.0b10...2.0.0b11](https://github.com/TheHive-Project/TheHive4py/compare/2.0.0b10...2.0.0b11)

tests/test_user_endpoint.py

@@ -10,10 +10,10 @@
 
 
 class TestUserEndpoint:
-    def test_get_current(self, thehive: TheHiveApi):
-        # TODO: implement a better control for user checking
+
+    def test_get_current(self, thehive: TheHiveApi, test_config: TestConfig):
         current_user = thehive.user.get_current()
-        assert current_user["login"] == "admin@thehive.local"
+        assert current_user["login"] == test_config.user
 
     def test_create_and_get(self, thehive: TheHiveApi):
         created_user = thehive.user.create(
@@ -47,11 +47,13 @@ def test_update(
     def test_lock_and_unlock(self, thehive: TheHiveApi, test_user: OutputUser):
         user_id = test_user["_id"]
 
-        thehive.user.lock(user_id=user_id)
+        with pytest.deprecated_call():
+            thehive.user.lock(user_id=user_id)
         locked_user = thehive.user.get(user_id=user_id)
         assert locked_user["locked"] is True
 
-        thehive.user.unlock(user_id=user_id)
+        with pytest.deprecated_call():
+            thehive.user.unlock(user_id=user_id)
         unlocked_user = thehive.user.get(user_id=user_id)
         assert unlocked_user["locked"] is False
 
@@ -61,7 +63,6 @@ def test_delete(self, thehive: TheHiveApi, test_user: OutputUser):
         with pytest.raises(TheHiveError):
             thehive.user.get(user_id=user_id)
 
-    @pytest.mark.skip(reason="integrator container only supports a single org ")
     def test_set_organisations(
         self, test_config: TestConfig, thehive: TheHiveApi, test_user: OutputUser
     ):
@@ -73,24 +74,41 @@ def test_set_organisations(
             },
             {
                 "default": False,
-                "organisation": test_config.share_org,
-                "profile": "read-only",
+                "organisation": test_config.admin_org,
+                "profile": "admin",
             },
         ]
         user_organisations = thehive.user.set_organisations(
             user_id=test_user["_id"], organisations=organisations
         )
         assert organisations == user_organisations
 
-    def test_set_password(self, thehive: TheHiveApi, test_user: OutputUser):
+    def test_set_and_change_password(self, thehive: TheHiveApi, test_user: OutputUser):
+
         assert test_user["hasPassword"] is False
         user_id = test_user["_id"]
 
         password = "super-secruht!"
         thehive.user.set_password(user_id=user_id, password=password)
 
-        user_with_password = thehive.user.get(user_id=user_id)
-        assert user_with_password["hasPassword"] is True
+        # apparently the change_password endpoint only works for the current user
+        # meaning no other user can invoke a password change for other users
+        thehive_test_user = TheHiveApi(
+            url=thehive.session.hive_url, username=test_user["login"], password=password
+        )
+
+        user_with_password = thehive_test_user.user.get_current()
+        assert user_with_password
+
+        new_password = "l4m0u|2t0uj0u|25"
+        thehive_test_user.user.change_password(
+            user_id=user_with_password["login"],
+            password=new_password,
+            current_password=password,
+        )
+
+        with pytest.raises(TheHiveError, match="AuthenticationError"):
+            thehive_test_user.user.get_current()
 
     def test_renew_get_and_remove_apikey(
         self, thehive: TheHiveApi, test_user: OutputUser

thehive4py/endpoints/user.py

@@ -1,3 +1,4 @@
+import warnings
 from typing import List, Optional
 
 from thehive4py.endpoints._base import EndpointBase
@@ -15,70 +16,218 @@
 
 
 class UserEndpoint(EndpointBase):
+    def get_current(self) -> OutputUser:
+        """Get the current session's user.
+
+        Returns:
+            The current session user.
+        """
+        return self._session.make_request("GET", path="/api/v1/user/current")
+
     def create(self, user: InputUser) -> OutputUser:
+        """Create a user.
+
+        Args:
+            user: The body of the user.
+
+        Returns:
+            The created user.
+        """
         return self._session.make_request("POST", path="/api/v1/user", json=user)
 
     def get(self, user_id: str) -> OutputUser:
+        """Get a user by id.
+
+        Args:
+            user_id: The id of the user.
+
+        Returns:
+            The user specified by the id.
+        """
         return self._session.make_request("GET", path=f"/api/v1/user/{user_id}")
 
-    def get_current(self) -> OutputUser:
-        return self._session.make_request("GET", path="/api/v1/user/current")
+    def lock(self, user_id: str) -> None:
+        """Lock a user.
 
-    def delete(self, user_id: str, organisation: Optional[str] = None) -> None:
-        return self._session.make_request(
-            "DELETE",
-            path=f"/api/v1/user/{user_id}/force",
-            params={"organisation": organisation},
+        !!! warning
+            Deprecated: use the generic [user.update]
+            [thehive4py.endpoints.user.UserEndpoint.update] method
+            to set the `locked` field to `True`
+
+        Args:
+            user_id: The id of the user.
+
+        Returns:
+            N/A
+        """
+        warnings.warn(
+            message=(
+                "Deprecated: use the generic user.update method to "
+                "set the `locked` field to True"
+            ),
+            category=DeprecationWarning,
+            stacklevel=2,
         )
+        return self.update(user_id=user_id, fields={"locked": True})
+
+    def unlock(self, user_id: str) -> None:
+        """Unlock a user.
+
+        !!! warning
+            Deprecated: use the generic [user.update]
+            [thehive4py.endpoints.user.UserEndpoint.update] method
+            to set the `locked` field to `False`
+
+        Args:
+            user_id: The id of the user.
+
+        Returns:
+            N/A
+        """
+        warnings.warn(
+            message=(
+                "Deprecated: use the generic user.update method to "
+                "set the `locked` field to False"
+            ),
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
+        return self.update(user_id=user_id, fields={"locked": False})
 
     def update(self, user_id: str, fields: InputUpdateUser) -> None:
+        """Update a user.
+
+        Args:
+            user_id: The id of the user.
+            fields: The fields of the user to update.
+
+        Returns:
+            N/A
+        """
         return self._session.make_request(
             "PATCH", path=f"/api/v1/user/{user_id}", json=fields
         )
 
-    def lock(self, user_id: str) -> None:
-        return self.update(user_id=user_id, fields={"locked": True})
+    def delete(self, user_id: str, organisation: Optional[str] = None) -> None:
+        """Delete a user.
 
-    def unlock(self, user_id: str) -> None:
-        return self.update(user_id=user_id, fields={"locked": False})
+        Args:
+            user_id: The id of the user.
+            organisation: The organisation from which to delete the user from. Optional.
+
+        Returns:
+            N/A
+        """
+        return self._session.make_request(
+            "DELETE",
+            path=f"/api/v1/user/{user_id}/force",
+            params={"organisation": organisation},
+        )
 
     def set_organisations(
         self, user_id: str, organisations: List[InputUserOrganisation]
     ) -> List[OutputUserOrganisation]:
+        """Set the organisations of a user.
+
+        Args:
+            user_id: The id of the user.
+            organisations: The list of organisations to set to the user.
+
+        Returns:
+            The list of the set user organisations.
+        """
         return self._session.make_request(
             "PUT",
             path=f"/api/v1/user/{user_id}/organisations",
             json={"organisations": organisations},
         )["organisations"]
 
     def set_password(self, user_id: str, password: str) -> None:
+        """Set the password of a user.
+
+        Args:
+            user_id: The id of the user.
+            password: The new password of the user.
+
+        Returns:
+            N/A
+        """
         return self._session.make_request(
             "POST",
             path=f"/api/v1/user/{user_id}/password/set",
             json={"password": password},
         )
 
+    def change_password(
+        self, user_id: str, password: str, current_password: str
+    ) -> None:
+        """Change the password of a user.
+
+        Args:
+            user_id: The id of the user.
+            password: The new password of the user.
+            current_password: The old password of the user.
+
+        Returns:
+            N/A
+        """
+        return self._session.make_request(
+            "POST",
+            path=f"/api/v1/user/{user_id}/password/change",
+            json={"password": password, "currentPassword": current_password},
+        )
+
     def get_apikey(self, user_id: str) -> str:
+        """Get the apikey of a user.
+
+        Args:
+            user_id: The id of the user.
+
+        Returns:
+            The apikey of the user.
+        """
         return self._session.make_request("GET", path=f"/api/v1/user/{user_id}/key")
 
     def remove_apikey(self, user_id: str) -> None:
+        """Remove the apikey of a user.
+
+        Args:
+            user_id: The id of the user.
+
+        Returns:
+            N/A
+        """
         return self._session.make_request("DELETE", path=f"/api/v1/user/{user_id}/key")
 
     def renew_apikey(self, user_id: str) -> str:
+        """Renew the apikey of a user.
+
+        Args:
+            user_id: The id of the user.
+
+        Returns:
+            The renewed apikey of the user.
+        """
         return self._session.make_request(
             "POST", path=f"/api/v1/user/{user_id}/key/renew"
         )
 
-    def get_avatar(self, user_id: str):
-        # TODO: implement the avatar download
-        raise NotImplementedError()
-
     def find(
         self,
         filters: Optional[FilterExpr] = None,
         sortby: Optional[SortExpr] = None,
         paginate: Optional[Paginate] = None,
     ) -> List[OutputUser]:
+        """Find multiple users.
+
+        Args:
+            filters: The filter expressions to apply in the query.
+            sortby: The sort expressions to apply in the query.
+            paginate: The pagination experssion to apply in the query.
+
+        Returns:
+            The list of users matched by the query or an empty list.
+        """
         query: QueryExpr = [
             {"_name": "listUser"},
             *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
@@ -92,6 +241,14 @@ def find(
         )
 
     def count(self, filters: Optional[FilterExpr] = None) -> int:
+        """Count users.
+
+        Args:
+            filters: The filter expressions to apply in the query.
+
+        Returns:
+            The count of users matched by the query.
+        """
         query: QueryExpr = [
             {"_name": "listUser"},
             *self._build_subquery(filters=filters),

thehive4py/types/user.py

@@ -1,4 +1,6 @@
-from typing import List, TypedDict
+from typing import List, Literal, TypedDict
+
+InputUserType = Literal["Normal", "Service"]
 
 
 class InputUserRequired(TypedDict):
@@ -11,15 +13,30 @@ class InputUser(InputUserRequired, total=False):
     email: str
     password: str
     organisation: str
-    type: str
+    type: InputUserType
+
+
+class OrganisationLinkRequired(TypedDict):
+    toOrganisation: str
+    linkType: str
+    otherLinkType: str
+
+
+class OrganisationLink(OrganisationLinkRequired, total=False):
+    avatar: str
 
 
-class OutputOrganisationProfile(TypedDict):
+class OutputOrganisationProfileRequired(TypedDict):
     organisationId: str
     organisation: str
     profile: str
 
 
+class OutputOrganisationProfile(OutputOrganisationProfileRequired, total=False):
+    avatar: str
+    links: List[OrganisationLink]
+
+
 class OutputUserRequired(TypedDict):
     _id: str
     _createdBy: str
@@ -54,6 +71,7 @@ class InputUpdateUser(TypedDict, total=False):
     avatar: str
     email: str
     defaultOrganisation: str
+    type: InputUserType
 
 
 class InputUserOrganisationRequired(TypedDict):