path: root/ipalib/plugins/aci.py

# Authors:
#   Rob Crittenden <rcritten@redhat.com>
#
# Copyright (C) 2009  Red Hat
# see file 'COPYING' for use and warranty information
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License as
# published by the Free Software Foundation; version 2 only
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA

"""
Frontend plugins for managing DS ACIs
"""

from ipalib import api, crud, errors2
from ipalib import Object, Command  # Plugin base classes
from ipalib import Str, Flag, Int  # Parameter types
from ipalib.aci import ACI

def make_aci(current, aciname, kw):
    try:
        taskgroup = api.Command['taskgroup_show'](kw['taskgroup'])
    except errors2.NotFound:
        # The task group doesn't exist, let's be helpful and add it
        tgkw = {'description':aciname}
        taskgroup = api.Command['taskgroup_add'](kw['taskgroup'], **tgkw)

    a = ACI(current)
    a.name = aciname
    a.permissions = kw['permissions'].replace(' ','').split(',')
    a.set_bindrule("groupdn = \"ldap:///%s\"" % taskgroup['dn'])
    if kw.get('attrs', None):
        a.set_target_attr(kw['attrs'].split())
    if kw.get('type', None):
        a.set_target_attr(kw['attrs'].split())
    if kw.get('memberof', None):
        group = api.Command['group_show'](kw['memberof'])
        a.set_target_filter("memberOf=%s" % group['dn'].decode('UTF-8'))
    return a

def search_by_name(acis, aciname):
    """
    Find an aci using the name field.

    Must be an exact match of the entire name.
    """
    for a in acis:
        try:
            t = ACI(a)
            if t.name == aciname:
                return str(t)
        except SyntaxError, e:
            # FIXME: need to log syntax errors, ignore for now
            pass

    raise errors2.NotFound()

def search_by_attr(acis, attrlist):
    """
    Find an aci by targetattr.

    Returns an ACI list of all acis the attribute appears in.
    """
    results = []
    for a in acis:
        try:
            t = ACI(a)
            for attr in attrlist:
                attr = attr.lower()
                for v in t.target['targetattr'].get('expression'):
                    if attr == v.lower():
                        results.append(str(t))
        except SyntaxError, e:
            # FIXME: need to log syntax errors, ignore for now
            pass

    if results:
        return results

    raise errors2.NotFound()

def search_by_taskgroup(acis, tgdn):
    """
    Find an aci by taskgroup. This searches the ACI bind rule.

    Returns an ACI list of all acis that match.
    """
    results = []
    for a in acis:
        try:
            t = ACI(a)
            if t.bindrule['expression'] == "ldap:///" + tgdn:
                results.append(str(t))
        except SyntaxError, e:
            # FIXME: need to log syntax errors, ignore for now
            pass

    if results:
        return results

    raise errors2.NotFound()

def search_by_perm(acis, permlist):
    """
    Find an aci by permissions

    Returns an ACI list of all acis the permission appears in.
    """
    results = []
    for a in acis:
        try:
            t = ACI(a)
            for perm in permlist:
                if perm.lower() in t.permissions:
                    results.append(str(t))
        except SyntaxError, e:
            # FIXME: need to log syntax errors, ignore for now
            pass

    if results:
        return results

    raise errors2.NotFound()

def search_by_memberof(acis, memberoffilter):
    """
    Find an aci by memberof

    Returns an ACI list of all acis that has a matching memberOf as a
    targetfilter.
    """
    results = []
    memberoffilter = memberoffilter.lower()
    for a in acis:
        try:
            t = ACI(a)
            try:
                if memberoffilter == t.target['targetfilter'].get('expression').lower():
                    results.append(str(t))
            except KeyError:
                pass
        except SyntaxError, e:
            # FIXME: need to log syntax errors, ignore for now
            pass

    if results:
        return results

    raise errors2.NotFound()

class aci(Object):
    """
    ACI object.
    """
    takes_params = (
        Str('aciname',
            doc='Name of ACI',
            primary_key=True,
        ),
        Str('taskgroup',
            doc='Name of taskgroup this ACI grants access to',
        ),
        Str('permissions',
            doc='Permissions to grant: read, write',
        ),
        Str('attrs?',
            doc='Comma-separated list of attributes',
        ),
        Str('type?',
            doc='type of IPA object: user, group, host',
        ),
        Str('memberof?',
            doc='member of a group',
        ),
        Str('filter?',
            doc='A legal LDAP filter (ou=Engineering)',
        ),
        Str('subtree?',
            doc='A subtree to apply the ACI to',
        ),
    )
api.register(aci)


class aci_add(crud.Create):
    """
    Add a new aci.
    """

    def execute(self, aciname, **kw):
        """
        Execute the aci-add operation.

        Returns the entry as it will be created in LDAP.

        :param aciname: The name of the ACI being added.
        :param kw: Keyword arguments for the other LDAP attributes.
        """
        assert 'aciname' not in kw
        ldap = self.api.Backend.ldap

        newaci = make_aci(None, aciname, kw)

        currentaci = ldap.retrieve(self.api.env.basedn, ['aci'])

        acilist = currentaci.get('aci')
        for a in acilist:
            try:
                b = ACI(a)
                if newaci.isequal(b):
                    raise errors2.DuplicateEntry()
            except SyntaxError:
                pass
        acilist.append(str(newaci))
        kwupdate = {'aci': acilist}

        return ldap.update(currentaci.get('dn'), **kwupdate)

api.register(aci_add)


class aci_del(crud.Delete):
    'Delete an existing aci.'
    """
    Remove an aci by name.
    """

    def execute(self, aciname, **kw):
        """
        Execute the aci-del operation.

        :param aciname: The name of the ACI being added.
        :param kw: unused
        """
        assert 'aciname' not in kw
        ldap = self.api.Backend.ldap

        currentaci = ldap.retrieve(self.api.env.basedn, ['aci'])
        acilist = currentaci.get('aci')
        a = search_by_name(acilist, aciname)
        i = acilist.index(str(a))
        del acilist[i]

        kwupdate = {'aci': acilist}

        return ldap.update(currentaci.get('dn'), **kwupdate)

    def output_for_cli(self, textui, result, aciname):
        """
        Output result of this command to command line interface.
        """
        textui.print_plain('Deleted aci "%s"' % aciname)

api.register(aci_del)


class aci_mod(crud.Update):
    'Edit an existing aci.'
    def execute(self, aciname, **kw):
        return "Not implemented"
    def output_for_cli(self, textui, result, aciname, **options):
        textui.print_plain(result)
api.register(aci_mod)


class aci_find(crud.Search):
    'Search for a aci.'
    takes_options = (
        Str('bindrule?',
            doc='The bindrule (e.g. ldap:///self)'
        ),
        Flag('and?',
             doc='Consider multiple options to be \"and\" so all are required.')
    )
    def execute(self, term, **kw):
        ldap = self.api.Backend.ldap
        currentaci = ldap.retrieve(self.api.env.basedn, ['aci'])
        currentaci = currentaci.get('aci')
        results = []

        # aciname
        if kw.get('aciname'):
            try:
                a = search_by_name(currentaci, kw.get('aciname'))
                results = [a]
                if kw.get('and'):
                    currentaci = results
            except errors2.NotFound:
                if kw.get('and'):
                    results = []
                    currentaci = []
                pass

        # attributes
        if kw.get('attrs'):
            try:
                attrs = kw.get('attrs')
                attrs = attrs.replace(' ','').split(',')
                a=search_by_attr(currentaci, attrs)
                if kw.get('and'):
                    results = a
                    currentaci = results
                else:
                    results = results + a
            except errors2.NotFound:
                if kw.get('and'):
                    results = []
                    currentaci = []
                pass

        # taskgroup
        if kw.get('taskgroup'):
            try:
                tg = api.Command['taskgroup_show'](kw.get('taskgroup'))
            except errors2.NotFound:
                # FIXME, need more precise error
                raise
            try:
                a=search_by_taskgroup(currentaci, tg.get('dn'))
                if kw.get('and'):
                    results = a
                    currentaci = results
                else:
                    results = results + a
            except errors2.NotFound:
                if kw.get('and'):
                    results = []
                    currentaci = []
                pass

        # permissions
        if kw.get('permissions'):
            try:
                permissions = kw.get('permissions')
                permissions = permissions.replace(' ','').split(',')
                a=search_by_perm(currentaci, permissions)
                if kw.get('and'):
                    results = a
                    currentaci = results
                else:
                    results = results + a
            except errors2.NotFound:
                if kw.get('and'):
                    results = []
                    currentaci = []
                pass

        # memberOf
        if kw.get('memberof'):
            try:
                group = api.Command['group_show'](kw['memberof'])
                memberof = "(memberOf=%s)" % group['dn'].decode('UTF-8')
                a=search_by_memberof(currentaci, memberof)
                results = results + a
                if kw.get('and'):
                    currentaci = results
            except errors2.NotFound:
                if kw.get('and'):
                    results = []
                    currentaci = []
                pass

# TODO
#  --type=STR         type of IPA object: user, group, host
#  --filter=STR       A legal LDAP filter (ou=Engineering)
#  --subtree=STR      A subtree to apply the ACI to
#  --bindrule=STR      A subtree to apply the ACI to

        # Make sure we have no dupes in the list
        results = list(set(results))

        # the first entry contains the count
        counter = len(results)
        return [counter] + results

    def output_for_cli(self, textui, result, term, **options):
        counter = result[0]
        acis = result[1:]
        if counter == 0 or len(acis) == 0:
            textui.print_plain("No entries found")
            return
        textui.print_name(self.name)
        for a in acis:
            textui.print_plain(a)
        textui.print_count(acis, '%d acis matched')

api.register(aci_find)


class aci_show(crud.Retrieve):
    'Examine an existing aci.'
    def execute(self, aciname, **kw):
        """
        Execute the aci-show operation.

        Returns the entry

        :param uid: The login name of the user to retrieve.
        :param kw: unused
        """
        ldap = self.api.Backend.ldap
        currentaci = ldap.retrieve(self.api.env.basedn, ['aci'])

        a = search_by_name(currentaci.get('aci'), aciname)
        return str(a)

    def output_for_cli(self, textui, result, aciname, **options):
        textui.print_plain(result)

api.register(aci_show)


class aci_showall(Command):
    'Examine all existing acis.'
    def execute(self):
        """
        Execute the aci-show operation.

        Returns the entry

        :param uid: The login name of the user to retrieve.
        :param kw: unused
        """
        ldap = self.api.Backend.ldap
        return ldap.retrieve(self.api.env.basedn, ['aci'])
    def output_for_cli(self, textui, result, **options):
        textui.print_entry(result)

api.register(aci_showall)