diff --git a/atlassian/crowd.py b/atlassian/crowd.py index 23912ef82..7627fd00d 100644 --- a/atlassian/crowd.py +++ b/atlassian/crowd.py @@ -59,15 +59,24 @@ def _user_change_status(self, username, active): data=user_object, ) - def user(self, username): + def user(self, username=None, key=None): """ - Get user information - :param username: + Get user information by name or by key + :param username: str - username, ignored when key is given + :param key: str - user key :return: """ - params = {"username": username} + params = {"key": key} if key else {"username": username} return self.get(self._crowd_api_url("usermanagement", "user"), params=params) + def user_by_openid(self, openid): + """ + Look up a user by the v2 OpenID URL. + :param openid: str - the v2 OpenID URL for the user + :return: user info + """ + return self.get(self._crowd_api_url("usermanagement", "user/id"), params={"openid": openid}) + def user_activate(self, username): """ Activate user @@ -131,27 +140,37 @@ def user_delete(self, username): return self.delete(self._crowd_api_url("usermanagement", "user"), params=params) - def user_groups(self, username, kind="direct"): + def user_groups(self, username, kind="direct", groupname=None, start_index=0, max_results=99999): """ Get user's all group info :param username: str - username :param kind: str - group type + :param groupname: str - optional single group name to retrieve + :param start_index: int - start index for paging + :param max_results: int - maximum number of results :return: The specify user's group info """ path = self._crowd_api_url("usermanagement", f"user/group/{kind}") - response = self.get(path, params={"username": username}) + params = {"username": username, "start-index": start_index, "max-results": max_results} + if groupname: + params["groupname"] = groupname + response = self.get(path, params=params) return search("groups[*].name", response) - def group_members(self, group, kind="direct", max_results=99999): + def group_members(self, group, kind="direct", username=None, start_index=0, max_results=99999): """ Get group's all direct members :param group: str - group name :param kind: str - group type + :param username: str - optional single username to retrieve + :param start_index: int - start index for paging :param max_results: int - maximum number of results :return: The specify group's direct members info """ path = self._crowd_api_url("usermanagement", f"group/user/{kind}") - params = {"groupname": group, "max-results": max_results} + params = {"groupname": group, "start-index": start_index, "max-results": max_results} + if username: + params["username"] = username response = self.get(path, params=params) return search("users[*].name", response) @@ -210,7 +229,7 @@ def user_attributes(self, username): def user_store_attributes(self, username, attributes): """Store user attributes using Crowd's attribute request body.""" - return self.put( + return self.post( self._crowd_api_url("usermanagement", "user/attribute"), params={"username": username}, data=attributes ) @@ -239,7 +258,7 @@ def group_attributes(self, groupname): def group_store_attributes(self, groupname, attributes): """Store group attributes using Crowd's attribute request body.""" - return self.put( + return self.post( self._crowd_api_url("usermanagement", "group/attribute"), params={"groupname": groupname}, data=attributes ) @@ -250,17 +269,34 @@ def group_remove_attribute(self, groupname, attribute_name): params={"groupname": groupname, "attributename": attribute_name}, ) - def nested_group_members(self, groupname, max_results=99999): - """Return users who are direct or nested members of a group.""" - response = self.get( - self._crowd_api_url("usermanagement", "group/user/nested"), - params={"groupname": groupname, "max-results": max_results}, - ) + def nested_group_members(self, groupname, username=None, start_index=0, max_results=99999): + """ + Return users who are direct or nested members of a group. + :param groupname: str - group name + :param username: str - optional single username to retrieve + :param start_index: int - start index for paging + :param max_results: int - maximum number of results + :return: list of usernames + """ + params = {"groupname": groupname, "start-index": start_index, "max-results": max_results} + if username: + params["username"] = username + response = self.get(self._crowd_api_url("usermanagement", "group/user/nested"), params=params) return search("users[*].name", response) - def nested_user_groups(self, username): - """Return direct and nested groups for a user.""" - response = self.get(self._crowd_api_url("usermanagement", "user/group/nested"), params={"username": username}) + def nested_user_groups(self, username, groupname=None, start_index=0, max_results=99999): + """ + Return direct and nested groups for a user. + :param username: str - username + :param groupname: str - optional single group name to retrieve + :param start_index: int - start index for paging + :param max_results: int - maximum number of results + :return: list of group names + """ + params = {"username": username, "start-index": start_index, "max-results": max_results} + if groupname: + params["groupname"] = groupname + response = self.get(self._crowd_api_url("usermanagement", "user/group/nested"), params=params) return search("groups[*].name", response) def group_child_groups(self, groupname, child_groupname=None, start_index=0, max_results=99999): @@ -852,6 +888,50 @@ def update_access_based_synchronization(self, application_id, data): """ return self.put(self._admin_api_url(f"application/{application_id}/access-based-synchronization"), data=data) + def get_email_scan_result(self, application_id): + """ + Get the latest duplicated email addresses scan result for an application. + :param application_id: str - application id + :return: scan result + """ + return self.get(self._admin_api_url(f"application/{application_id}/emailscan")) + + def trigger_email_scan(self, application_id): + """ + Trigger a duplicated email addresses scan for an application. + :param application_id: str - application id + :return: + """ + return self.post(self._admin_api_url(f"application/{application_id}/emailscan")) + + def dismiss_message(self, message_key): + """ + Dismiss an administration console message. + :param message_key: str - message key + :return: + """ + return self.post(self._admin_api_url(f"dismiss/{message_key}")) + + def get_encryption_settings(self): + """Get database encryption settings.""" + return self.get(self._admin_api_url("encryption")) + + def change_encryption_key(self): + """Change the key used by the current encryptor.""" + return self.put(self._admin_api_url("encryption/changeKey")) + + def disable_encryption(self): + """Disable database encryption.""" + return self.put(self._admin_api_url("encryption/disable")) + + def set_default_encryptor(self, data): + """ + Set the default encryptor. + :param data: dict - encryptor representation, e.g. {"key": "AES"} + :return: + """ + return self.put(self._admin_api_url("encryption/encryptor"), data=data) + def get_directory_mappings(self, application_id, start=0, limit=99999): """ Get directory mappings for an application. @@ -1429,8 +1509,8 @@ def find_saml_directory_mapping_mismatch(self, application_id): return self.get(self._admin_api_url(f"samlconfig/application/{application_id}/directory-mapping-mismatch")) def get_dynamic_ldap_pool_statistics(self): - """Get dynamic LDAP pool statistics.""" - return self.get(self._admin_api_url("dynamic-ldap-pool-statistics")) + """Get dynamic LDAP connection pool statistics.""" + return self.get(self._admin_api_url("spring-ldap-pool-statistics")) def save_mail_configuration(self, data): """ diff --git a/docs/crowd.rst b/docs/crowd.rst index 8ac9cc4d4..57ea9bddc 100644 --- a/docs/crowd.rst +++ b/docs/crowd.rst @@ -25,11 +25,15 @@ Manage users # Delete user crowd.user_delete(username) - # Get user + # Get user by name or by key crowd.user(username) + crowd.user(key="557057:927441f1-cc92-4030-b633-8a2bbdf7136e") + + # Get user by the v2 OpenID URL + crowd.user_by_openid("https://crowd.example.test/openidserver/users/ada") # Get user's all group info - crowd.user_groups(username, kind='direct') + crowd.user_groups(username, kind='direct', groupname=None, start_index=0, max_results=99999) # Check whether the user is a member of the group crowd.is_user_in_group(username, group, kind='direct') @@ -37,6 +41,7 @@ Manage users crowd.user_update(username, {"name": username, "email": "ada@example.com", "active": True}) crowd.user_update_password(username, "new-password") crowd.user_delete_password(username) + crowd.user_attributes(username) crowd.user_store_attributes(username, {"attributes": []}) crowd.user_remove_attribute(username, "attribute-name") crowd.nested_user_groups(username) @@ -58,13 +63,14 @@ Manage groups crowd.group_add_user(username, groupname) # Get group's members - crowd.group_members(group, kind='direct', max_results=99999) + crowd.group_members(group, kind='direct', username=None, start_index=0, max_results=99999) # Create new group method crowd.group_create(groupname, description, active=True) crowd.group(groupname) crowd.group_update(groupname, {"name": groupname, "active": True}) + crowd.group_attributes(groupname) crowd.group_store_attributes(groupname, {"attributes": []}) crowd.group_remove_attribute(groupname, "attribute-name") crowd.nested_group_members(groupname) @@ -189,6 +195,10 @@ Administration crowd.update_directory_mapping("application-id", "directory-id", {"directoryId": "dir1"}) crowd.delete_directory_mapping("application-id", "directory-id") crowd.move_directory_mapping("application-id", "directory-id", {"position": 1}) + crowd.get_access_based_synchronization("application-id") + crowd.update_access_based_synchronization("application-id", {"filterType": "USER_ONLY_FILTERING"}) + crowd.get_email_scan_result("application-id") + crowd.trigger_email_scan("application-id") # Directories crowd.get_detailed_directories(active=True) @@ -258,6 +268,7 @@ Administration crowd.get_saml_application_config("application-id") crowd.update_saml_application_config("application-id", {"enabled": True}) crowd.parse_saml_metadata("...") + crowd.parse_saml_metadata_file("/path/to/metadata.xml") crowd.reset_saml_certificates() crowd.get_saml_idp_metadata() crowd.find_saml_directory_mapping_mismatch("application-id") @@ -268,6 +279,15 @@ Administration crowd.validate_mail_configuration({"host": "smtp.example.com"}) crowd.get_dynamic_ldap_pool_statistics() + # Database encryption + crowd.get_encryption_settings() + crowd.set_default_encryptor({"key": "AES"}) + crowd.change_encryption_key() + crowd.disable_encryption() + + # Console messages + crowd.dismiss_message("message-key") + Healthcheck ------------- diff --git a/tests/crowd/test_crowd.py b/tests/crowd/test_crowd.py index ebf0a15d9..7a2603d75 100644 --- a/tests/crowd/test_crowd.py +++ b/tests/crowd/test_crowd.py @@ -13,11 +13,9 @@ def test_crowd_user_group_and_attribute_helpers(mock_put, mock_get, mock_delete) assert crowd.nested_group_members("engineering") == ["ada"] crowd.group_remove_user("ada", "engineering") crowd.user_update_password("ada", "new-secret") - crowd.group_store_attributes("engineering", {"attributes": []}) assert mock_delete.call_args_list[0].kwargs["params"] == {"username": "ada", "groupname": "engineering"} assert mock_put.call_args_list[0].kwargs["data"] == {"value": "new-secret"} - assert mock_put.call_args_list[1].kwargs["params"] == {"groupname": "engineering"} @patch.object(Crowd, "post") @@ -412,4 +410,99 @@ def test_admin_saml_and_ldap_methods(mock_post, mock_get): assert mock_post.call_args.kwargs["headers"] == {"Content-Type": "application/octet-stream"} crowd.get_dynamic_ldap_pool_statistics() - assert mock_get.call_args.args[0].endswith("admin/1.0/dynamic-ldap-pool-statistics") + assert mock_get.call_args.args[0].endswith("admin/1.0/spring-ldap-pool-statistics") + + +@patch.object(Crowd, "get") +def test_user_lookup_by_key_and_openid(mock_get): + crowd = Crowd("https://crowd.example.test", "application", "password") + + crowd.user("ada") + assert mock_get.call_args.kwargs["params"] == {"username": "ada"} + + crowd.user(key="557057:927441f1") + assert mock_get.call_args.kwargs["params"] == {"key": "557057:927441f1"} + + crowd.user_by_openid("https://crowd.example.test/openidserver/users/ada") + assert mock_get.call_args.args[0].endswith("usermanagement/latest/user/id") + assert mock_get.call_args.kwargs["params"] == {"openid": "https://crowd.example.test/openidserver/users/ada"} + + +@patch.object(Crowd, "get") +def test_membership_lookups_support_paging_and_single_entity_filters(mock_get): + crowd = Crowd("https://crowd.example.test", "application", "password") + mock_get.return_value = {"groups": [{"name": "engineering"}], "users": [{"name": "ada"}]} + + assert crowd.user_groups("ada", groupname="engineering", start_index=10, max_results=20) == ["engineering"] + assert mock_get.call_args.kwargs["params"] == { + "username": "ada", + "groupname": "engineering", + "start-index": 10, + "max-results": 20, + } + + assert crowd.group_members("engineering", username="ada", start_index=5) == ["ada"] + assert mock_get.call_args.kwargs["params"]["username"] == "ada" + assert mock_get.call_args.kwargs["params"]["start-index"] == 5 + + crowd.nested_group_members("engineering", username="ada") + assert mock_get.call_args.kwargs["params"]["username"] == "ada" + + crowd.nested_user_groups("ada", groupname="engineering") + assert mock_get.call_args.args[0].endswith("usermanagement/latest/user/group/nested") + assert mock_get.call_args.kwargs["params"]["groupname"] == "engineering" + + +@patch.object(Crowd, "post") +def test_store_attributes_uses_post(mock_post): + crowd = Crowd("https://crowd.example.test", "application", "password") + + crowd.user_store_attributes("ada", {"attributes": []}) + assert mock_post.call_args.args[0].endswith("usermanagement/latest/user/attribute") + assert mock_post.call_args.kwargs["params"] == {"username": "ada"} + + crowd.group_store_attributes("engineering", {"attributes": []}) + assert mock_post.call_args.args[0].endswith("usermanagement/latest/group/attribute") + assert mock_post.call_args.kwargs["params"] == {"groupname": "engineering"} + + +@patch.object(Crowd, "put") +@patch.object(Crowd, "post") +@patch.object(Crowd, "get") +def test_admin_encryption_email_scan_and_dismiss_methods(mock_get, mock_post, mock_put): + crowd = Crowd("https://crowd.example.test", "application", "password") + + crowd.get_email_scan_result("app1") + assert mock_get.call_args.args[0].endswith("admin/1.0/application/app1/emailscan") + + crowd.trigger_email_scan("app1") + assert mock_post.call_args.args[0].endswith("admin/1.0/application/app1/emailscan") + + crowd.dismiss_message("upgrade.message") + assert mock_post.call_args.args[0].endswith("admin/1.0/dismiss/upgrade.message") + + crowd.get_encryption_settings() + assert mock_get.call_args.args[0].endswith("admin/1.0/encryption") + + crowd.set_default_encryptor({"key": "AES"}) + assert mock_put.call_args.args[0].endswith("admin/1.0/encryption/encryptor") + assert mock_put.call_args.kwargs["data"] == {"key": "AES"} + + crowd.change_encryption_key() + assert mock_put.call_args.args[0].endswith("admin/1.0/encryption/changeKey") + + crowd.disable_encryption() + assert mock_put.call_args.args[0].endswith("admin/1.0/encryption/disable") + + +@patch.object(Crowd, "put") +@patch.object(Crowd, "get") +def test_admin_access_based_synchronization_methods(mock_get, mock_put): + crowd = Crowd("https://crowd.example.test", "application", "password") + + crowd.get_access_based_synchronization("app1") + assert mock_get.call_args.args[0].endswith("admin/1.0/application/app1/access-based-synchronization") + + crowd.update_access_based_synchronization("app1", {"filterType": "USER_ONLY_FILTERING"}) + assert mock_put.call_args.args[0].endswith("admin/1.0/application/app1/access-based-synchronization") + assert mock_put.call_args.kwargs["data"] == {"filterType": "USER_ONLY_FILTERING"}