workgroup.py

# Copyright 2016
#      The Board of Trustees of the Leland Stanford Junior University
#      Stanford Research Computing Center
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
Stanford University Workgroup Python bindings

Module to manage workgroups through the Restful Web Service API

manager = SUWorkgroupManager("https://workgroupsvc.stanford.edu/v1",
                             CERT, CERTKEY)

workgroup = manager.workgroup(name)
workgroup = manager.create_workgroup(name)
manager.delete_workgroup(name)

READ ONLY GETTERS
-----------------
workgroup.members -> list of sunetid string
workgroup.applications -> list of application id
workgroup.workgroups -> list of workgroup objects
workgroup.administrators -> list of sunetid strings
workgroup.admin_apps -> list of admin app id
workgroup.admin_workgroups -> list of workgroup objects

READ/WRITE PROPERTIES
-----------=======----
workgroup.description (string)
workgroup.filter (string)
workgroup.reusable (boolean)
workgroup.visibility (string)
workgroup.privgroup (boolean)

EDIT MEMBERS
------------
workgroup.add_member('sthiell')
workgroup.add_workgroup(child_workgroup)

workgroup.remove_member('sthiell')
workgroup.remove_workgroup(child_workgroup)

EDIT ADMINS
-----------
workgroup.add_administrator('sthiell')
workgroup.add_admin_workgroup(admin_workgroup)

workgroup.remove_administrator('sthiell')
workgroup.remove_admin_workgroup(admin_workgroup)
"""

__author__ = 'sthiell@stanford.edu (Stephane Thiell)'

import requests
from textwrap import dedent
import xml.etree.ElementTree as ET
from xml.parsers.expat import ExpatError


DEFAULT_BASE_URL = "https://workgroupsvc.stanford.edu/v1"

# Workgroup filter property
FILTER_NONE = 'NONE'
FILTER_ACADEMIC_ADMINISTRATIVE = 'ACADEMIC_ADMINISTRATIVE'
FILTER_STUDENT = 'STUDENT'
FILTER_FACULTY = 'FACULTY'
FILTER_STAFF = 'STAFF'
FILTER_FACULTY_STAFF = 'FACULTY_STAFF'
FILTER_FACULTY_STUDENT = 'FACULTY_STUDENT'
FILTER_STAFF_STUDENT = 'STAFF_STUDENT'
FILTER_FACULTY_STAFF_STUDENT = 'FACULTY_STAFF_STUDENT'

# Workgroup visibility property
VISIBILITY_STANFORD = 'STANFORD'
VISIBILITY_PRIVATE = 'PRIVATE'


class SUWorkgroupAPIError(Exception):
    """SU Workgroup API Error Exception"""

    def __init__(self, status_code, text):
        """initialize API error exception"""
        self.status_code = status_code
        oneline = '|'.join([line.strip() for line in text.splitlines()])
        Exception.__init__(self, oneline)

    def __str__(self):
        return "Error %d: %s" % (self.status_code, self.args[0])


class SUWorkgroup(object):
    """Stanford University Workgroup

    Provides accessors and workgroup modifiers.

    Do not instantiate directly: use SUWorkgroupManager.workgroup().
    """

    def __init__(self, name, manager):
        """Initializer - called by SUWorkgroupManager"""
        # workgroup properties
        self.name = name
        self._description = ''
        self._filter = FILTER_NONE
        self._visibility = VISIBILITY_STANFORD
        self._reusable = True
        self._privgroup = False

        # workgroup members
        self.members = set()
        self.applications = set()
        self.workgroups = {}

        # workgroup administrators
        self.administrators = set()
        self.admin_apps = set()
        self.admin_workgroups = {}

        # workgroup manager (SUWorkgroupManager)
        self.manager = manager

    @property
    def description(self):
        """Workgroup description property (string)."""
        return self._description

    @description.setter
    def description(self, description):
        """Update workgroup description property (string)."""
        self.manager._request_update_property(self.name, 'description',
                                              description)
        self._description = description

    @property
    def filter(self):
        """Workgroup filter property (string)."""
        return self._filter

    @filter.setter
    def filter(self, filter):
        """Update workgroup filter property (string)."""
        self.manager._request_update_property(self.name, 'filter', filter)
        self._filter = filter

    @property
    def reusable(self):
        """Workgroup reusable property (boolean)."""
        return self._reusable

    @reusable.setter
    def reusable(self, reusable):
        """Update workgroup reusable property (boolean)."""
        if reusable:
            reusable_str = 'TRUE'
        else:
            reusable_str = 'FALSE'
        self.manager._request_update_property(self.name, 'reusable',
                                              reusable_str)
        self._reusable = bool(reusable)

    @property
    def visibility(self):
        """Workgroup visibility property (string)."""
        return self._visibility

    @visibility.setter
    def visibility(self, visibility):
        """Update workgroup visibility property (string)."""
        self.manager._request_update_property(self.name, 'visibility',
                                              visibility)
        self._visibility = visibility

    @property
    def privgroup(self):
        """Workgroup privgroup property (boolean)."""
        return self._privgroup

    @privgroup.setter
    def privgroup(self, privgroup):
        """Update workgroup privgroup property (boolean)."""
        if privgroup:
            privgroup_str = 'TRUE'
        else:
            privgroup_str = 'FALSE'
        self.manager._request_update_property(self.name, 'privgroup',
                                              privgroup_str)
        self._privgroup = bool(privgroup)

    def add_member(self, member):
        """Add an user (string) as member of workgroup object."""
        self.manager._request('put', self.name, member,
                              nested_workgroup=False, admin=False)
        self.members.add(member)

    def remove_member(self, member):
        """Remove an member user (string) from workgroup object."""
        self.manager._request('delete', self.name, member,
                              nested_workgroup=False, admin=False)
        self.members.remove(member)

    def add_administrator(self, admin_member):
        """Add an user (string) as administrator of workgroup object."""
        self.manager._request('put', self.name, admin_member,
                              nested_workgroup=False, admin=True)
        self.administrators.add(admin_member)

    def remove_administrator(self, admin_member):
        """Remove an administrator (string) from workgroup object."""
        self.manager._request('delete', self.name, admin_member,
                              nested_workgroup=False, admin=True)
        self.administrators.remove(admin_member)

    def add_workgroup(self, workgroup):
        """
        Add workgroup as a member of workgroup object (nested workgroup).

        workgroup should be a valid SUWorkgroup object
        """
        self.manager._request('put', self.name, workgroup.name,
                              nested_workgroup=True, admin=False)
        self.workgroups[workgroup.name] = workgroup

    def remove_workgroup(self, workgroup):
        """
        Remove a workgroup member from workgroup object (nested workgroup).

        workgroup should be a valid SUWorkgroup object
        """
        self.manager._request('delete', self.name, workgroup.name,
                              nested_workgroup=True, admin=False)
        del self.workgroups[workgroup.name]

    def add_admin_workgroup(self, workgroup):
        """
        Add workgroup as an admin of workgroup object.
        workgroup may be a valid SUWorkgroup object or a string.
        """
        self.manager._request('put', self.name, workgroup.name,
                              nested_workgroup=True, admin=True)
        self.admin_workgroups[workgroup.name] = workgroup

    def remove_admin_workgroup(self, workgroup):
        """
        Remove admin workgroup from workgroup object.
        workgroup may be a valid SUWorkgroup object or a string.
        """
        self.manager._request('delete', self.name, workgroup.name,
                              nested_workgroup=True, admin=True)
        del self.admin_workgroups[workgroup.name]


class SUWorkgroupManager(object):
    """Stanford University Workgroup Manager class

    Class used to get and operate on workgroups.

    Starting
    --------
    Create SUWorkgroupManager object with proper base_url and cert info.

    Getting an existing workgroup
    -----------------------------
    Use the workgroup(name) method to get an SUWorkgroup object, you may
    then access workgroup data or modify workgroup data using SUWorkgroup
    methods.

    Creating a new workgroup
    ------------------------
    Use create_workgroup(name, ...) to create a new workgroup. This method
    also returns a new SUWorkgroup object that you can use.

    Deleting a workgroup
    --------------------
    Use delete_workgroup(name) to delete a workgroup. At the time of writing,
    it is not possible to reuse a deleted workgroup.

    Deleting cached workgroups
    --------------------------
    This class currently caches workgroups. use clear() to clear caches.
    """

    def __init__(self, base_url=DEFAULT_BASE_URL, cert=None, cert_key=None):
        """Initializer, you probably want to pass valid cert/cert_key info."""
        self.base_url = base_url
        self.cert = cert
        self.cert_key = cert_key
        self.workgroups = {} # cached workgroups

    def clear(self):
        """Clear cached workgroups."""
        self.workgroups = {}

    def _request_workgroup(self, workgroup_name, depth=0):
        """load"""
        if workgroup_name in self.workgroups:
            # cache hit
            return self.workgroups[workgroup_name]

        res = requests.get("%s/workgroups/%s" % (self.base_url, workgroup_name),
                           verify=False, cert=(self.cert, self.cert_key))
        self._check_response(res)

        # 200 OK
        root = ET.fromstring(res.text)

        workgroup = SUWorkgroup(workgroup_name, self)
        self.workgroups[workgroup_name] = workgroup

        for child in root:
            if child.tag == "description":
                workgroup._description = child.text
            elif child.tag == "filter":
                workgroup._filter = child.text or FILTER_NONE
            elif child.tag == "visibility":
                workgroup._visibility = child.text
            elif child.tag == "reusable":
                workgroup._reusable = (child.text == 'TRUE')
            elif child.tag == "privgroup":
                workgroup._privgroup = (child.text == 'TRUE')
            elif child.tag == "members":
                for member in child:
                    if member.tag == "workgroup":
                        child_name = member.attrib['id']
                        child_workgroup = self._request_workgroup(child_name,
                                                                  depth + 1)
                        assert child_name not in workgroup.workgroups
                        workgroup.workgroups[child_name] = child_workgroup
                    elif member.tag == "member":
                        workgroup.members.add(member.attrib['id'])
                    elif member.tag == "application":
                        workgroup.applications.add(member.attrib['id'])
            elif child.tag == "administrators":
                for member in child:
                    if member.tag == "workgroup":
                        child_name = member.attrib['id']
                        child_workgroup = self._request_workgroup(child_name,
                                                                  depth + 1)
                        assert child_name not in workgroup.admin_workgroups
                        workgroup.admin_workgroups[child_name] = child_workgroup
                    elif member.tag == "member":
                        workgroup.administrators.add(member.attrib['id'])
                    elif member.tag == "application":
                        workgroup.admin_apps.add(member.attrib['id'])

        return workgroup

    def _request(self, reqtype, workgroup_name, member, nested_workgroup=False,
                 admin=False):
        """
        Generic Request for Workgroup operation (put, delete)
        reqtype is the request type like put, delete (match requests.xx calls)
        workgroup_name is the name of workgroup we operate on
        member is the name of the new member
        nested_workgroup: whether the member is a workgroup
        admin: whether to add new member as an admin of workgroup
        """
        if nested_workgroup:
            member_objtype = 'workgroups'
        else:
            member_objtype = 'users'

        if admin:
            section = 'administrators'
            query_keyword = 'administrator'
        else:
            section = 'members'
            query_keyword = 'user'

        payload = {query_keyword: '%s/%s/%s' % (self.base_url,
                                                member_objtype,
                                                member)}

        reqfunc = getattr(requests, reqtype)
        res = reqfunc("%s/workgroups/%s/%s" % (self.base_url,
                                               workgroup_name,
                                               section),
                      verify=False, cert=(self.cert, self.cert_key),
                      params=payload)
        self._check_response(res)

    def _request_update_property(self, workgroup_name, property, value):
        """
        Generic Request for property change on a workgroup (description,
        filter, etc.)

        workgroup_name is the name of workgroup we operate on property is
        the name of the workgroup property to update value is the new
        workgroup property value
        """
        headers = {'Content-Type': 'text/plain;charset=UTF-8'}
        res = requests.put("%s/workgroups/%s/%s" % (self.base_url,
                                                    workgroup_name,
                                                    property),
                           verify=False, cert=(self.cert, self.cert_key),
                           data=value, headers=headers)
        self._check_response(res)

    def _check_response(self, response):
        """Check HTTP response"""
        code = response.status_code
        errmsg = ''
        try:
            if len(response.text) > 0:
                root = ET.fromstring(response.text)
                if root.tag == "error":
                    for child in root:
                        if child.tag == "message":
                            errmsg = child.text
        except ExpatError:
            pass

        if code >= 300 or errmsg:
            raise SUWorkgroupAPIError(code, errmsg)

    def workgroup(self, name):
        """Get workgroup object from given workgroup name."""
        assert name
        if name not in self.workgroups:
            return self._request_workgroup(name)
        return self.workgroups[name]

    def create_workgroup(self, name, description, filter=FILTER_NONE,
                         visibility=VISIBILITY_STANFORD, reusable=True,
                         privgroup=True):
        """Create new workgroup object with given name."""
        assert name
        assert description
        payloadfmt = dedent("""
            <workgroup>
                <description>%s</description>
                <filter>%s</filter>
                <visibility>%s</visibility>
                <reusable>%s</reusable>
                <privgroup>%s</privgroup>
            </workgroup>""")
        payload = payloadfmt % (description, filter, visibility,
                                str(reusable).upper(),
                                str(privgroup).upper())
        headers = {'Content-Type': 'text/xml;charset=UTF-8'}
        res = requests.post("%s/workgroups/%s" % (self.base_url, name),
                            verify=False, cert=(self.cert, self.cert_key),
                            data=payload, headers=headers)

        if res.status_code not in (requests.codes.ok, requests.codes.created):
            raise SUWorkgroupAPIError(res.status_code, res.text)

        return self.workgroup(name)

    def delete_workgroup(self, name):
        """Delete workgroup object of given name."""
        assert name
        res = requests.delete("%s/workgroups/%s" % (self.base_url, name),
                              verify=False, cert=(self.cert, self.cert_key))

        if res.status_code != requests.codes.ok:
            raise SUWorkgroupAPIError(res.status_code, res.text)