Add new password policy plugin based on baseldap.py classes.

2 files changed, 522 insertions, 0 deletions

diff --git a/ipalib/plugins/pwpolicy2.py b/ipalib/plugins/pwpolicy2.py
new file mode 100644
index 00000000..797c0810
--- /dev/null
+++ b/ipalib/plugins/pwpolicy2.py
@@ -0,0 +1,351 @@
+# Authors:
+#   Pavel Zuna <pzuna@redhat.com>
+#
+# Copyright (C) 2010  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
+"""
+Password policy
+"""
+
+from ipalib import api
+from ipalib import Int, Str
+from ipalib.plugins.baseldap import *
+from ipalib import _
+
+
+class cosentry(LDAPObject):
+    """
+    Class of Service object used for linking policies with groups
+    """
+    INTERNAL = True
+
+    container_dn = 'cn=costemplates,%s' % api.env.container_accounts
+    object_class = ['top', 'costemplate', 'extensibleobject', 'krbcontainer']
+    default_attributes = ['cn', 'cospriority', 'krbpwdpolicyreference']
+
+    takes_params = (
+        Str('cn', primary_key=True),
+        Str('krbpwdpolicyreference'),
+        Int('cospriority', minvalue=0),
+    )
+
+    priority_not_unique_msg = _(
+        'priority must be a unique value (%(prio)d already used by %(gname)s)'
+    )
+
+    def get_dn(self, *keys, **options):
+        group_dn = self.api.Object.group.get_dn(keys[-1])
+        return self.backend.make_dn_from_attr(
+            'cn', group_dn, self.container_dn
+        )
+
+    def check_priority_uniqueness(self, *keys, **options):
+        if options.get('cospriority') is not None:
+            entries = self.methods.find(
+                cospriority=options['cospriority']
+            )['result']
+            if len(entries) > 0:
+                group_name = self.api.Object.group.get_primary_key_from_dn(
+                    entries[0]['cn'][0]
+                )
+                raise errors.ValidationError(
+                    name='priority',
+                    error=self.priority_not_unique_msg % {
+                        'prio': options['cospriority'],
+                        'gname': group_name,
+                    }
+                )
+
+api.register(cosentry)
+
+
+class cosentry_add(LDAPCreate):
+    INTERNAL = True
+
+    def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys, **options):
+        # check for existence of the group
+        self.api.Command.group_show(keys[-1])
+        self.obj.check_priority_uniqueness(*keys, **options)
+        del entry_attrs['cn']
+        return dn
+
+api.register(cosentry_add)
+
+
+class cosentry_del(LDAPDelete):
+    INTERNAL = True
+
+api.register(cosentry_del)
+
+
+class cosentry_mod(LDAPUpdate):
+    INTERNAL = True
+
+    def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys, **options):
+        self.obj.check_priority_uniqueness(*keys, **options)
+        return dn
+
+api.register(cosentry_mod)
+
+
+class cosentry_show(LDAPRetrieve):
+    INTERNAL = True
+
+api.register(cosentry_show)
+
+
+class cosentry_find(LDAPSearch):
+    INTERNAL = True
+
+api.register(cosentry_find)
+
+
+GLOBAL_POLICY_NAME = u'GLOBAL'
+
+
+class pwpolicy2(LDAPObject):
+    """
+    Password Policy object
+    """
+    container_dn = 'cn=%s,cn=kerberos' % api.env.realm
+    object_name = 'password policy'
+    object_name_plural = 'password policies'
+    object_class = ['top', 'nscontainer', 'krbpwdpolicy']
+    default_attributes = [
+        'cn', 'cospriority', 'krbmaxpwdlife', 'krbminpwdlife',
+        'krbpwdhistorylength', 'krbpwdmindiffchars', 'krbpwdminlength',
+    ]
+
+    takes_params = (
+        Str('cn?',
+            cli_name='group',
+            label=_('Group'),
+            doc=_('Manage password policy for specific group'),
+            primary_key=True,
+        ),
+        Int('krbmaxpwdlife?',
+            cli_name='maxlife',
+            label=_('Max lifetime (days)'),
+            doc=_('Maximum password lifetime (in days)'),
+            minvalue=0,
+        ),
+        Int('krbminpwdlife?',
+            cli_name='minlife',
+            label=_('Min lifetime (hours)'),
+            doc=_('Minimum password lifetime (in hours)'),
+            minvalue=0,
+        ),
+        Int('krbpwdhistorylength?',
+            cli_name='history',
+            label=_('History size'),
+            doc=_('Password history size'),
+            minvalue=0,
+        ),
+        Int('krbpwdmindiffchars?',
+            cli_name='minclasses',
+            label=_('Character classes'),
+            doc=_('Minimum number of character classes'),
+            minvalue=0,
+            maxvalue=5,
+        ),
+        Int('krbpwdminlength?',
+            cli_name='minlength',
+            label=_('Min length'),
+            doc=_('Minimum length of password'),
+            minvalue=0,
+        ),
+        Int('cospriority',
+            cli_name='priority',
+            label=_('Priority'),
+            doc=_('Priority of the policy (higher number means lower priority'),
+            minvalue=0,
+        ),
+    )
+
+    def get_dn(self, *keys, **options):
+        if keys[-1] is not None:
+            return self.backend.make_dn_from_attr(
+                self.primary_key.name, keys[-1], self.container_dn
+            )
+        return self.api.env.container_accounts
+
+    def convert_time_for_output(self, entry_attrs, **options):
+        # Convert seconds to hours and days for displaying to user
+        if not options.get('raw', False):
+            if 'krbmaxpwdlife' in entry_attrs:
+                entry_attrs['krbmaxpwdlife'][0] = unicode(
+                    int(entry_attrs['krbmaxpwdlife'][0]) / 86400
+                )
+            if 'krbminpwdlife' in entry_attrs:
+                entry_attrs['krbminpwdlife'][0] = unicode(
+                    int(entry_attrs['krbminpwdlife'][0]) / 3600
+                )
+
+    def convert_time_on_input(self, entry_attrs):
+        # Convert hours and days to seconds for writing to LDAP
+        if 'krbmaxpwdlife' in entry_attrs and entry_attrs['krbmaxpwdlife']:
+            entry_attrs['krbmaxpwdlife'] = entry_attrs['krbmaxpwdlife'] * 86400
+        if 'krbminpwdlife' in entry_attrs and entry_attrs['krbminpwdlife']:
+            entry_attrs['krbminpwdlife'] = entry_attrs['krbminpwdlife'] * 3600
+
+api.register(pwpolicy2)
+
+
+class pwpolicy2_add(LDAPCreate):
+    """
+    Create new group password policy.
+    """
+    def get_args(self):
+        yield self.obj.primary_key.clone(attribute=True, required=True)
+
+    def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys, **options):
+        self.api.Command.cosentry_add(
+            keys[-1], krbpwdpolicyreference=dn,
+            cospriority=options.get('cospriority')
+        )
+        if 'cospriority' in entry_attrs:
+            del entry_attrs['cospriority']
+        self.obj.convert_time_on_input(entry_attrs)
+        return dn
+
+    def post_callback(self, ldap, dn, entry_attrs, *keys, **options):
+        self.log.info('%r' % entry_attrs)
+        if not options.get('raw', False):
+            if options.get('cospriority') is not None:
+                entry_attrs['cospriority'] = [unicode(options['cospriority'])]
+        self.obj.convert_time_for_output(entry_attrs, **options)
+        return dn
+
+api.register(pwpolicy2_add)
+
+
+class pwpolicy2_del(LDAPDelete):
+    """
+    Delete group password policy.
+    """
+    def get_args(self):
+        yield self.obj.primary_key.clone(attribute=True, required=True)
+
+    def post_callback(self, ldap, dn, *keys, **options):
+        try:
+            self.api.Command.cosentry_del(keys[-1])
+        except errors.NotFound:
+            pass
+        return True
+
+api.register(pwpolicy2_del)
+
+
+class pwpolicy2_mod(LDAPUpdate):
+    """
+    Modify group password policy.
+    """
+    def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys, **options):
+        if options.get('cospriority') is not None:
+            if keys[-1] is None:
+                raise errors.ValidationError(
+                    name='priority',
+                    error=_('priority cannot be set on global policy')
+                )
+            try:
+                self.api.Command.cosentry_mod(
+                    keys[-1], cospriority=options['cospriority']
+                )
+            except errors.NotFound:
+                self.api.Command.cosentry_add(
+                    keys[-1], krbpwdpolicyreference=dn,
+                    cospriority=options['cospriority']
+                )
+            del entry_attrs['cospriority']
+        self.obj.convert_time_on_input(entry_attrs)
+        return dn
+
+    def post_callback(self, ldap, dn, entry_attrs, *keys, **options):
+        if not options.get('raw', False):
+            if options.get('cospriority') is not None:
+                entry_attrs['cospriority'] = [unicode(options['copriority'])]
+            if keys[-1] is None:
+                entry_attrs['cn'] = GLOBAL_POLICY_NAME
+        self.obj.convert_time_for_output(entry_attrs, **options)
+        return dn
+
+api.register(pwpolicy2_mod)
+
+
+class pwpolicy2_show(LDAPRetrieve):
+    """
+    Display group password policy.
+    """
+    takes_options = (
+        Str('user?',
+            label=_('User'),
+            doc=_('Display effective policy for a specific user'),
+        ),
+    )
+
+    def pre_callback(self, ldap, dn, attrs_list, *keys, **options):
+        if options.get('user') is not None:
+            user_entry = self.api.Command.user_show(
+                options['user'], all=True
+            )['result']
+            if 'krbpwdpolicyreference' in user_entry:
+                return user_entry.get('krbpwdpolicyreference', [dn])[0]
+        return dn
+
+    def post_callback(self, ldap, dn, entry_attrs, *keys, **options):
+        if not options.get('raw', False):
+            if keys[-1] is not None:
+                try:
+                    cos_entry = self.api.Command.cosentry_show(
+                        keys[-1]
+                    )['result']
+                    if cos_entry.get('cospriority') is not None:
+                        entry_attrs['cospriority'] = cos_entry['cospriority']
+                except errors.NotFound:
+                    pass
+            else:
+                entry_attrs['cn'] = GLOBAL_POLICY_NAME
+        self.obj.convert_time_for_output(entry_attrs, **options)
+        return dn
+
+api.register(pwpolicy2_show)
+
+
+class pwpolicy2_find(LDAPSearch):
+    """
+    Search for group password policies.
+    """
+    def post_callback(self, ldap, entries, truncated, *args, **options):
+        if not options.get('raw', False):
+            for e in entries:
+                try:
+                    cos_entry = self.api.Command.cosentry_show(
+                        e[1]['cn'][0]
+                    )['result']
+                    if cos_entry.get('cospriority') is not None:
+                        e[1]['cospriority'] = cos_entry['cospriority']
+                except errors.NotFound:
+                    pass
+                self.obj.convert_time_for_output(e[1], **options)
+        global_entry = self.api.Command.pwpolicy2_show(
+            all=options.get('all', False), raw=options.get('raw', False)
+        )['result']
+        dn = global_entry['dn']
+        del global_entry['dn']
+        entries.insert(0, (dn, global_entry))
+
+api.register(pwpolicy2_find)
+
diff --git a/tests/test_xmlrpc/test_pwpolicy2.py b/tests/test_xmlrpc/test_pwpolicy2.py
new file mode 100644
index 00000000..72c65bed
--- /dev/null
+++ b/tests/test_xmlrpc/test_pwpolicy2.py
@@ -0,0 +1,171 @@
+# Authors:
+#   Rob Crittenden <rcritten@redhat.com>
+#   Pavel Zuna <pzuna@redhat.com>
+#
+# Copyright (C) 2010  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
+"""
+Test the `ipalib/plugins/pwpolicy2.py` module.
+"""
+
+import sys
+from xmlrpc_test import XMLRPC_test, assert_attr_equal
+from ipalib import api
+from ipalib import errors
+
+
+class test_pwpolicy2(XMLRPC_test):
+    """
+    Test the `pwpolicy2` plugin.
+    """
+    group = u'testgroup12'
+    group2 = u'testgroup22'
+    user = u'testuser12'
+    kw = {'cospriority': 1, 'krbminpwdlife': 30, 'krbmaxpwdlife': 40, 'krbpwdhistorylength': 5, 'krbpwdminlength': 6 }
+    kw2 = {'cospriority': 2, 'krbminpwdlife': 40, 'krbmaxpwdlife': 60, 'krbpwdhistorylength': 8, 'krbpwdminlength': 9 }
+
+    def test_1_pwpolicy2_add(self):
+        """
+        Test adding a per-group policy using the `xmlrpc.pwpolicy2_add` method.
+        """
+        # First set up a group and user that will use this policy
+        self.failsafe_add(
+            api.Object.group, self.group, description=u'pwpolicy test group',
+        )
+        self.failsafe_add(
+            api.Object.user, self.user, givenname=u'Test', sn=u'User'
+        )
+        api.Command.group_add_member(self.group, users=self.user)
+
+        entry = api.Command['pwpolicy2_add'](self.group, **self.kw)['result']
+        assert_attr_equal(entry, 'krbminpwdlife', '30')
+        assert_attr_equal(entry, 'krbmaxpwdlife', '40')
+        assert_attr_equal(entry, 'krbpwdhistorylength', '5')
+        assert_attr_equal(entry, 'krbpwdminlength', '6')
+        assert_attr_equal(entry, 'cospriority', '1')
+
+    def test_2_pwpolicy2_add(self):
+        """
+        Add a policy with a already used priority.
+
+        The priority validation is done first, so it's OK that the group
+        is the same here.
+        """
+        try:
+            api.Command['pwpolicy2_add'](self.group, **self.kw)
+        except errors.ValidationError:
+            pass
+        else:
+            assert False
+
+    def test_3_pwpolicy2_add(self):
+        """
+        Add a policy that already exists.
+        """
+        try:
+            # cospriority needs to be unique
+            self.kw['cospriority'] = 3
+            api.Command['pwpolicy2_add'](self.group, **self.kw)
+        except errors.DuplicateEntry:
+            pass
+        else:
+            assert False
+
+    def test_4_pwpolicy2_add(self):
+        """
+        Test adding another per-group policy using the `xmlrpc.pwpolicy2_add` method.
+        """
+        self.failsafe_add(
+            api.Object.group, self.group2, description=u'pwpolicy test group 2'
+        )
+        entry = api.Command['pwpolicy2_add'](self.group2, **self.kw2)['result']
+        assert_attr_equal(entry, 'krbminpwdlife', '40')
+        assert_attr_equal(entry, 'krbmaxpwdlife', '60')
+        assert_attr_equal(entry, 'krbpwdhistorylength', '8')
+        assert_attr_equal(entry, 'krbpwdminlength', '9')
+        assert_attr_equal(entry, 'cospriority', '2')
+
+    def test_5_pwpolicy2_add(self):
+        """
+        Add a pwpolicy for a non-existent group
+        """
+        try:
+            api.Command['pwpolicy2_add'](u'nopwpolicy', cospriority=1, krbminpwdlife=1)
+        except errors.NotFound:
+            pass
+        else:
+            assert False
+
+    def test_6_pwpolicy2_show(self):
+        """
+        Test the `xmlrpc.pwpolicy2_show` method with global policy.
+        """
+        entry = api.Command['pwpolicy2_show']()['result']
+        # Note that this assumes an unchanged global policy
+        assert_attr_equal(entry, 'krbminpwdlife', '1')
+        assert_attr_equal(entry, 'krbmaxpwdlife', '90')
+        assert_attr_equal(entry, 'krbpwdhistorylength', '0')
+        assert_attr_equal(entry, 'krbpwdminlength', '8')
+
+    def test_7_pwpolicy2_show(self):
+        """
+        Test the `xmlrpc.pwpolicy2_show` method.
+        """
+        entry = api.Command['pwpolicy2_show'](self.group)['result']
+        assert_attr_equal(entry, 'krbminpwdlife', '30')
+        assert_attr_equal(entry, 'krbmaxpwdlife', '40')
+        assert_attr_equal(entry, 'krbpwdhistorylength', '5')
+        assert_attr_equal(entry, 'krbpwdminlength', '6')
+        assert_attr_equal(entry, 'cospriority', '1')
+
+    def test_8_pwpolicy2_mod(self):
+        """
+        Test the `xmlrpc.pwpolicy2_mod` method for global policy.
+        """
+        entry = api.Command['pwpolicy2_mod'](krbminpwdlife=50)['result']
+        assert_attr_equal(entry, 'krbminpwdlife', '50')
+
+        # Great, now change it back
+        entry = api.Command['pwpolicy2_mod'](krbminpwdlife=1)['result']
+        assert_attr_equal(entry, 'krbminpwdlife', '1')
+
+    def test_9_pwpolicy2_mod(self):
+        """
+        Test the `xmlrpc.pwpolicy2_mod` method.
+        """
+        entry = api.Command['pwpolicy2_mod'](self.group, krbminpwdlife=50)['result']
+        assert_attr_equal(entry, 'krbminpwdlife', '50')
+
+    def test_a_pwpolicy_del(self):
+        """
+        Test the `xmlrpc.pwpolicy2_del` method.
+        """
+        assert api.Command['pwpolicy2_del'](self.group)['result'] is True
+        # Verify that it is gone
+        try:
+            api.Command['pwpolicy2_show'](self.group)
+        except errors.NotFound:
+            pass
+        else:
+            assert False
+
+        # Remove the groups we created
+        api.Command['group_del'](self.group)
+        api.Command['group_del'](self.group2)
+
+        # Remove the user we created
+        api.Command['user_del'](self.user)
+