crowd.py

# coding=utf-8
import logging

from jmespath import search
from bs4 import BeautifulSoup

from .rest_client import AtlassianRestAPI

log = logging.getLogger(__name__)


class Crowd(AtlassianRestAPI):
    """Crowd API wrapper.
    Important to note that you will have to use an application credentials,
    not user credentials, in order to access Crowd APIs"""

    def __init__(
        self,
        url,
        username,
        password,
        timeout=60,
        api_root="rest",
        api_version="latest",
    ):
        super(Crowd, self).__init__(url, username, password, timeout, api_root, api_version)

    def _crowd_api_url(self, api, resource):
        return f"/{self.api_root}/{api}/{self.api_version}/{resource}"

    def _user_change_status(self, username, active):
        """
        Change user status.
        :param username: str - username
        :param active: bool - True/False
        :return:
        """

        user = self.user(username)

        user_object = {
            "name": username,
            "active": active,
            "display-name": user.get("display-name"),
            "first-name": user.get("first-name"),
            "last-name": user.get("last-name"),
            "email": user.get("email"),
        }

        params = {"username": username}

        return self.put(
            self._crowd_api_url("usermanagement", "user"),
            params=params,
            data=user_object,
        )

    def user(self, username):
        """
        Get user information
        :param username:
        :return:
        """
        params = {"username": username}
        return self.get(self._crowd_api_url("usermanagement", "user"), params=params)

    def user_activate(self, username):
        """
        Activate user
        :param username: str - username
        """

        return self._user_change_status(username, True)

    def user_create(
        self,
        username,
        active,
        first_name,
        last_name,
        display_name,
        email,
        password,
    ):
        """
        Create new user method
        :param  active: bool:
        :param  username: string: username
        :param  active: bool:
        :param  first_name: string:
        :param  last_name: string:
        :param  display_name:  string:
        :param  email: string:
        :param  password: string:
        :return:
        """

        user_object = {
            "name": username,
            "password": {"value": password},
            "active": active,
            "first-name": first_name,
            "last-name": last_name,
            "display-name": display_name,
            "email": email,
        }

        return self.post(self._crowd_api_url("usermanagement", "user"), data=user_object)

    def user_deactivate(self, username):
        """
        Deactivate user

        :return:
        """

        return self._user_change_status(username, False)

    def user_delete(self, username):
        """
        Delete user
        :param username: str - username
        :return:
        """

        params = {"username": username}

        return self.delete(self._crowd_api_url("usermanagement", "user"), params=params)

    def user_groups(self, username, kind="direct"):
        """
        Get user's all group info
        :param username: str - username
        :param kind: str - group type
        :return: The specify user's group info
        """
        path = self._crowd_api_url("usermanagement", f"user/group/{kind}")
        response = self.get(path, params={"username": username})
        return search("groups[*].name", response)

    def group_members(self, group, kind="direct", max_results=99999):
        """
        Get group's all direct members
        :param group: str - group name
        :param kind: str - group type
        :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}
        response = self.get(path, params=params)
        return search("users[*].name", response)

    def is_user_in_group(self, username, group, kind="direct"):
        """
        Check if the user is a member of the group
        :param username: str - username
        :param group: str - group name
        :param kind: str - group type
        :return: bool - Return `True` or `False`
        """
        path = self._crowd_api_url("usermanagement", f"group/user/{kind}")
        params = {"username": username, "groupname": group}
        response = self.get(path, params=params, advanced_mode=True)
        return response.status_code == 200

    def group_add_user(self, username, groupname):
        """
        Add user to group
        :return:
        """

        data = {"name": groupname}

        params = {"username": username}

        return self.post(
            self._crowd_api_url("usermanagement", "user/group/direct"),
            params=params,
            json=data,
        )

    def health_check(self):
        """
        Get health status
        https://confluence.atlassian.com/jirakb/how-to-retrieve-health-check-results-using-rest-api-867195158.html
        :return:
        """
        # check as Troubleshooting & Support Tools Plugin
        response = self.get("rest/troubleshooting/1.0/check/")
        if not response:
            # check as support tools
            response = self.get("rest/supportHealthCheck/1.0/check/")
        return response

    def get_plugins_info(self):
        """
        Provide plugins info
        :return a json of installed plugins
        """
        url = "rest/plugins/1.0/"
        return self.get(url, headers=self.no_check_headers, trailing=True)

    def get_plugin_info(self, plugin_key):
        """
        Provide plugin info
        :return a json of installed plugins
        """
        url = f"rest/plugins/1.0/{plugin_key}-key"
        return self.get(url, headers=self.no_check_headers, trailing=True)

    def get_plugin_license_info(self, plugin_key):
        """
        Provide plugin license info
        :return a json specific License query
        """
        url = f"rest/plugins/1.0/{plugin_key}-key/license"
        return self.get(url, headers=self.no_check_headers, trailing=True)

    def upload_plugin(self, plugin_path):
        """
        Provide plugin path for upload into Jira e.g. useful for auto deploy
        :param plugin_path:
        :return:
        """
        files = {"plugin": open(plugin_path, "rb")}
        upm_token = self.request(
            method="GET",
            path="rest/plugins/1.0/",
            headers=self.no_check_headers,
            trailing=True,
        ).headers["upm-token"]
        url = f"rest/plugins/1.0/?token={upm_token}"
        return self.post(url, files=files, headers=self.no_check_headers)

    def delete_plugin(self, plugin_key):
        """
        Delete plugin
        :param plugin_key:
        :return:
        """
        url = f"rest/plugins/1.0/{plugin_key}-key"
        return self.delete(url)

    def check_plugin_manager_status(self):
        url = "rest/plugins/latest/safe-mode"
        return self.request(method="GET", path=url, headers=self.safe_mode_headers)

    def update_plugin_license(self, plugin_key, raw_license):
        """
        Update license for plugin
        :param plugin_key:
        :param raw_license:
        :return:
        """
        app_headers = {
            "X-Atlassian-Token": "no-check",
            "Content-Type": "application/vnd.atl.plugins+json",
        }
        url = f"/plugins/1.0/{plugin_key}/license"
        data = {"rawLicense": raw_license}
        return self.put(url, data=data, headers=app_headers)

    @property
    def memberships(self):
        """
        Retrieves full details of all group memberships, with users and nested groups.
        See: https://docs.atlassian.com/atlassian-crowd/5.3.1/REST/#usermanagement/1/group-getAllMemberships
        :return: All membership mapping dict
        """
        path = self._crowd_api_url("usermanagement", "group/membership")
        headers = {"Accept": "application/xml"}
        response = self.get(path, headers=headers)
        soup = BeautifulSoup(response, "xml")
        memberships = {}
        for membership in soup.find_all("membership"):
            group = membership["group"]
            users = [user["name"] for user in membership.find_all("user")]
            memberships[group] = users
        return memberships

    def group_create(self, groupname, description=None, active=True):
        """
        Create new group method
        :param groupname: string: The name of new group
        :param description: string: The description of new group, default is None
        :param active: bool: Weather the group is active, default is True
        :return: Create result
        """
        group = {
            "name": groupname,
            "active": active,
            "description": description,
            "type": "GROUP",
        }

        return self.post(self._crowd_api_url("usermanagement", "group"), data=group)