Adding script bin/check-ldap-pwd-schemes and its application module pp_admintools...

diff --git a/bin/check-ldap-pwd-schemes b/bin/check-ldap-pwd-schemes
new file mode 100755 (executable)
index 0000000..6d89fdf
--- /dev/null
+++ b/bin/check-ldap-pwd-schemes
@@ -0,0 +1,75 @@
+#!/usr/bin/env python3
+"""
+@summary: Reports hashing schemes of userPassword attributes in LDAP.
+
+This application retrieves the password hashing schemes of userPassword
+attributes. Without a filter or a Base-DN the hashing schemes of all
+userPassword attributes are shown.
+
+One can filter the output by a regular LDAP-filter rule of by one or more
+hashing schemes.
+
+@author: Frank Brehm
+@contact: frank.brehm@pixelpark.com
+@copyright: © 2023 by Frank Brehm, Berlin
+"""
+from __future__ import print_function
+
+# Standard modules
+import locale
+import os
+import sys
+try:
+    from pathlib import Path
+except ImportError:
+    from pathlib2 import Path
+
+__exp_py_version_major__ = 3
+__min_py_version_minor__ = 6
+
+if sys.version_info[0] != __exp_py_version_major__:
+    print('This script is intended to use with Python {}.'.format(
+        __exp_py_version_major__), file=sys.stderr)
+    print('You are using Python: {0}.{1}.{2}-{3}-{4}.'.format(
+        *sys.version_info) + '\n', file=sys.stderr)
+    sys.exit(1)
+
+if sys.version_info[1] < __min_py_version_minor__:
+    print('A minimal Python version of {maj}.{min} is necessary to execute this script.'.format(
+        maj=__exp_py_version_major__, min=__min_py_version_minor__), file=sys.stderr)
+    print('You are using Python: {0}.{1}.{2}-{3}-{4}.'.format(
+        *sys.version_info) + '\n', file=sys.stderr)
+    sys.exit(1)
+
+__author__ = 'Frank Brehm <frank.brehm@pixelpark.com>'
+__copyright__ = '(C) 2023 by Frank Brehm, Digitas Pixelpark GmbH, Berlin'
+
+# own modules:
+
+my_path = Path(__file__)
+my_real_path = my_path.resolve()
+bin_path = my_real_path.parent
+base_dir = bin_path.parent
+lib_dir = base_dir.joinpath('lib')
+module_dir = lib_dir.joinpath('pp_admintools')
+
+if module_dir.exists():
+    sys.path.insert(0, str(lib_dir))
+
+from pp_admintools.app.check_ldap_pwd_schemes import CheckLdapPwdSchemesApplication
+
+appname = os.path.basename(sys.argv[0])
+
+locale.setlocale(locale.LC_ALL, '')
+
+app = CheckLdapPwdSchemesApplication(appname=appname, base_dir=base_dir)
+app.initialized = True
+
+if app.verbose > 2:
+    print('{c}-Object:\n{a}'.format(c=app.__class__.__name__, a=app))
+
+app()
+
+sys.exit(0)
+
+# vim: tabstop=4 expandtab shiftwidth=4 softtabstop=4

diff --git a/lib/pp_admintools/app/check_ldap_pwd_schemes.py b/lib/pp_admintools/app/check_ldap_pwd_schemes.py
new file mode 100644 (file)
index 0000000..6895f65
--- /dev/null
+++ b/lib/pp_admintools/app/check_ldap_pwd_schemes.py
@@ -0,0 +1,270 @@
+# -*- coding: utf-8 -*-
+"""
+@summary: Reports hashing schemes of userPassword attributes in LDAP.
+
+This application retrieves the password hashing schemes of userPassword
+attributes. Without a filter or a Base-DN the hashing schemes of all
+userPassword attributes are shown.
+
+One can filter the output by a regular LDAP-filter rule of by one or more
+hashing schemes.
+
+@author: Frank Brehm
+@contact: frank.brehm@pixelpark.com
+@copyright: © 2023 by Frank Brehm, Berlin
+"""
+from __future__ import absolute_import
+
+# Standard modules
+import logging
+
+# Third party modules
+
+# Own modules
+from fb_tools.collections import CIDict, CIStringSet
+from fb_tools.xlate import format_list
+
+from .ldap import BaseLdapApplication
+# from .ldap import LdapAppError, FatalLDAPError
+from .ldap import LdapAppError
+from .. import pp
+from ..config.ldap import LdapConfiguration
+from ..handler.ldap_password import LdapPasswordHandler
+from ..xlate import XLATOR
+
+__version__ = '0.1.0'
+LOG = logging.getLogger(__name__)
+
+_ = XLATOR.gettext
+ngettext = XLATOR.ngettext
+
+
+# =============================================================================
+class CheckLdapPwdSchemesError(LdapAppError):
+    """Special exception class for exceptions inside this module."""
+
+    pass
+
+
+# =============================================================================
+class CheckLdapPwdSchemesApplication(BaseLdapApplication):
+    """Application class for checking password hashing schemes in LDAP."""
+
+    show_simulate_option = False
+
+    use_default_ldap_connection = False
+    use_multiple_ldap_connections = False
+    show_cmdline_ldap_timeout = True
+    apply_default_ldap_instance_if_not_given = False
+    show_force_option = False
+    show_assume_options = False
+    default_filter = '(userPassword=*)(|(cn=*)(uid=*))'
+
+    # -------------------------------------------------------------------------
+    def __init__(self, appname=None, base_dir=None):
+        """Initialize the CheckLdapPwdSchemesApplication object."""
+        LdapPasswordHandler.init_pass_schemes()
+        self.ldap = None
+        self.instance = None
+        self.connect_info = None
+        self.add_ldap_filter = None
+        self.filter_schemes = []
+        self.show_details = False
+        self.found_dns = CIStringSet()
+        self.found_entries = CIDict()
+
+        self.get_attributes = CIStringSet()
+        self.get_attributes.add('cn')
+        self.get_attributes.add('uid')
+        self.get_attributes.add('userPassword')
+
+        self.pwd_handler = LdapPasswordHandler(
+            appname=appname, base_dir=base_dir, initialized=False)
+
+        desc = _(
+            'Reports the password hashing schemes of userPassword attributes in LDAP. '
+            'This application retrieves the password hashing schemes of userPassword '
+            'attributes. Without a filter or a Base-DN the hashing schemes of all '
+            'userPassword attributes are shown. One can filter the output by a regular '
+            'LDAP-filter rule of by one or more hashing schemes.')
+
+        super(CheckLdapPwdSchemesApplication, self).__init__(
+            appname=appname, description=desc, base_dir=base_dir,
+            cfg_class=LdapConfiguration, initialized=False)
+
+        self.pwd_handler.verbose = self.verbose
+
+        self.initialized = True
+
+    # -------------------------------------------------------------------------
+    def _verify_instances(self):
+
+        super(CheckLdapPwdSchemesApplication, self)._verify_instances(is_admin=True)
+
+    # -------------------------------------------------------------------------
+    def init_arg_parser(self):
+        """Initialze the argument parser with some special parameters."""
+        app_group = self.arg_parser.add_argument_group(_('Script options'))
+
+        app_group.add_argument(
+            '-F', '--filter', dest='filter', metavar=_('FILTER'),
+            help=_(
+                'An additional LDAP filter to limit the entries to print out their '
+                'password hashing schema. Please note, that this filter will be wrapped '
+                'by parantheses to combine it with the default '
+                'filter {!r}.').format(self.default_filter),
+        )
+
+        schema_list = []
+        for method in LdapPasswordHandler.available_schemes:
+            schema_id = LdapPasswordHandler.schema_ids[method]
+            schema_list.append(schema_id)
+        schema_list.append('list')
+        schema_list.append('help')
+
+        help_txt = _(
+            'One or more schemes (hashing methods) for filtering the userPassword attributes. '
+            'It is possible to give here the value {val_list!r}, then all possible schemes '
+            'are shown and exit.').format(val_list='list')
+
+        app_group.add_argument(
+            '-S', '--schema', metavar=_('SCHEMA'), dest='filter_schema', choices=schema_list,
+            nargs='*', help=help_txt
+        )
+
+        app_group.add_argument(
+            '-D', '--details', dest='details', action='store_true',
+            help=_('Show more details in output, e.g. the entry DN.'),
+        )
+
+        super(CheckLdapPwdSchemesApplication, self).init_arg_parser()
+
+    # -------------------------------------------------------------------------
+    def post_init(self):
+        """Execute some actions after initialising."""
+        super(CheckLdapPwdSchemesApplication, self).post_init()
+
+        self.export_file = getattr(self.args, 'export', None)
+        if self.export_file:
+            if not self.re_yaml_extension.search(self.export_file):
+                self.export_file += '.yaml'
+
+        self.add_ldap_filter = getattr(self.args, 'filter', None)
+        self.show_details = getattr(self.args, 'details', False)
+
+        if hasattr(self.args, 'filter_schema'):
+            schemes = getattr(self.args, 'filter_schema')
+            if schemes:
+                for schema in self.args.filter_schema:
+                    self.filter_schemes.append(schema)
+
+        self.instance = self.ldap_instances[0]
+        self.connect_info = self.cfg.ldap_connection[self.instance]
+
+    # -------------------------------------------------------------------------
+    def _run(self):
+
+        ldap_url = self.cfg.ldap_connection[self.instance].url
+
+        msg = _(
+            'Start reporting password hashing schemes if userPassword attributes '
+            'in in LDAP instance {inst!r} ({url}) ...').format(inst=self.instance, url=ldap_url)
+        LOG.debug(msg)
+
+        ldap_filter = self.default_filter
+        if self.add_ldap_filter:
+            ldap_filter = '(&' + self.default_filter + '(' + self.add_ldap_filter + '))'
+        else:
+            ldap_filter = '(&' + self.default_filter + ')'
+
+        LOG.debug(_('Used LDAP filter: {!r}.').format(ldap_filter))
+
+        for dn in self.get_all_entry_dns(self.instance, ldap_filter=ldap_filter):
+            self.perform_entry(dn)
+
+        if self.verbose > 3:
+            msg = _('Found entry DNs:') + '\n' + pp(self.found_entries.as_dict())
+            LOG.debug(msg)
+
+        max_len_name = 1
+        for dn in self.found_dns:
+            pwd_data = self.found_entries[dn]
+            used_name = pwd_data['name']
+            if len(used_name) > max_len_name:
+                max_len_name = len(used_name)
+
+        ml = max_len_name + 1
+
+        for dn in self.found_dns:
+            pwd_data = self.found_entries[dn]
+            n = pwd_data['name'] + ':'
+            schemes = format_list(pwd_data['hashing_methods'])
+
+            if self.show_details:
+                print()
+
+            print(f'{n:<{ml}} {schemes}')
+
+            if self.show_details:
+                print('    dn: ' + dn)
+
+    # -------------------------------------------------------------------------
+    def perform_entry(self, dn):
+        """Get the entry of the given DN and evaluate the password hashing method."""
+        if self.verbose > 1:
+            LOG.debug(_('Checking password hashing method of entry {!r} ...').format(dn))
+
+        get_attributes = self.get_attributes.as_list()
+
+        entry = self.get_entry(dn, self.instance, attributes=get_attributes)
+        attribs = self.normalized_attributes(entry)
+        if self.verbose > 2:
+            LOG.debug(_('Got attributes:') + '\n' + pp(attribs.as_dict()))
+
+        if 'userPassword' not in attribs:
+            return False
+
+        used_name = ''
+        if 'cn' in attribs and len(attribs['cn']):
+            used_name = attribs['cn'][0]
+        else:
+            used_name = attribs['uid'][0]
+
+        methods = []
+        show_entry = True
+        if self.filter_schemes:
+            show_entry = False
+        for passwd in attribs['userPassword']:
+            hash_method = self.pwd_handler.get_hashing_schema(passwd)
+            methods.append(hash_method)
+            if self.filter_schemes:
+                if hash_method in self.filter_schemes:
+                    show_entry = True
+
+        if show_entry and len(methods) > 0:
+            pwdata = {
+                'name': used_name,
+                'hashing_methods': methods,
+            }
+            if self.show_details:
+                pwdata['userPassword'] = attribs['userPassword']
+            self.found_dns.add(dn)
+            self.found_entries[dn] = pwdata
+
+            if self.verbose > 2:
+                msg = _('Found entry {!r} with data:').format(dn) + '\n' + pp(pwdata)
+                LOG.debug(msg)
+
+            return True
+
+        return False
+
+
+# =============================================================================
+if __name__ == '__main__':
+
+    pass
+
+# =============================================================================
+
+# vim: tabstop=4 expandtab shiftwidth=4 softtabstop=4 list