From 5287b9ac635a502f3cec9f6ac92f7f847b71d437 Mon Sep 17 00:00:00 2001 From: Simon Pichugin Date: Aug 26 2019 19:22:18 +0000 Subject: Issue 50206 - Refactor lock, unlock and status of dsidm account/role Description: Port ns-accountstatus.pl, ns-activate.pl and ns-inactivate.pl to lib389 CLI. Add: dsidm account/role entry-status, dsidm account subtree-status, dsidm role lock/unlock Refactor: dsidm account lock/unlock Remove: dsidm account status Also, refactor role.py and idm/account.py accordingly to the CLI requirements. https://pagure.io/389-ds-base/issue/50206 Reviewed by: firstyear (Thanks, William!) --- diff --git a/dirsrvtests/tests/suites/acl/acivattr_test.py b/dirsrvtests/tests/suites/acl/acivattr_test.py index f9cf096..e46b891 100644 --- a/dirsrvtests/tests/suites/acl/acivattr_test.py +++ b/dirsrvtests/tests/suites/acl/acivattr_test.py @@ -15,7 +15,7 @@ from lib389.cos import CosTemplate, CosClassicDefinition from lib389.topologies import topology_st as topo from lib389.idm.nscontainer import nsContainer from lib389.idm.domain import Domain -from lib389.idm.role import FilterRoles +from lib389.idm.role import FilteredRoles pytestmark = pytest.mark.tier1 @@ -55,7 +55,7 @@ def _add_user(request, topo): ou = OrganizationalUnit(topo.standalone, "ou=sales,o=acivattr,{}".format(DEFAULT_SUFFIX)) ou.create(properties={'ou': 'sales'}) - roles = FilterRoles(topo.standalone, DNBASE) + roles = FilteredRoles(topo.standalone, DNBASE) roles.create(properties={'cn':'FILTERROLEENGROLE', 'nsRoleFilter':'cn=eng*'}) roles.create(properties={'cn': 'FILTERROLESALESROLE', 'nsRoleFilter': 'cn=sales*'}) diff --git a/dirsrvtests/tests/suites/acl/roledn_test.py b/dirsrvtests/tests/suites/acl/roledn_test.py index f627753..3616532 100644 --- a/dirsrvtests/tests/suites/acl/roledn_test.py +++ b/dirsrvtests/tests/suites/acl/roledn_test.py @@ -18,7 +18,7 @@ from lib389.idm.user import UserAccounts, UserAccount from lib389.idm.organizationalunit import OrganizationalUnits from lib389.topologies import topology_st as topo from lib389.idm.domain import Domain -from lib389.idm.role import NestedRoles, ManagedRoles, FilterRoles +from lib389.idm.role import NestedRoles, ManagedRoles, FilteredRoles from lib389.idm.account import Anonymous import ldap @@ -94,7 +94,7 @@ def _add_user(request, topo): for i in ['ROLE1', 'ROLE21', 'ROLE31']: managedroles.create(properties={'cn': i}) - filterroles = FilterRoles(topo.standalone, OU_ROLE) + filterroles = FilteredRoles(topo.standalone, OU_ROLE) filterroles.create(properties={'cn': 'filterRole', 'nsRoleFilter': 'sn=Dr Drake', 'description': 'filter role tester'}) diff --git a/dirsrvtests/tests/suites/cos/cos_test.py b/dirsrvtests/tests/suites/cos/cos_test.py index a8907d1..6065e89 100644 --- a/dirsrvtests/tests/suites/cos/cos_test.py +++ b/dirsrvtests/tests/suites/cos/cos_test.py @@ -10,7 +10,7 @@ import pytest, os, ldap from lib389.cos import CosClassicDefinition, CosClassicDefinitions, CosTemplate from lib389._constants import DEFAULT_SUFFIX from lib389.topologies import topology_st as topo -from lib389.idm.role import FilterRoles +from lib389.idm.role import FilteredRoles from lib389.idm.nscontainer import nsContainer from lib389.idm.user import UserAccount @@ -36,7 +36,7 @@ def test_positive(topo): 6. Operation should success """ # Adding ns filter role - roles = FilterRoles(topo.standalone, DEFAULT_SUFFIX) + roles = FilteredRoles(topo.standalone, DEFAULT_SUFFIX) roles.create(properties={'cn': 'FILTERROLEENGROLE', 'nsRoleFilter': 'cn=eng*'}) # adding ns container diff --git a/dirsrvtests/tests/suites/filter/vfilter_simple_test.py b/dirsrvtests/tests/suites/filter/vfilter_simple_test.py index 871fd99..1796f88 100644 --- a/dirsrvtests/tests/suites/filter/vfilter_simple_test.py +++ b/dirsrvtests/tests/suites/filter/vfilter_simple_test.py @@ -19,7 +19,7 @@ from lib389.idm.organizationalunit import OrganizationalUnits from lib389.idm.account import Accounts from lib389.idm.user import UserAccount, UserAccounts from lib389.schema import Schema -from lib389.idm.role import ManagedRoles, FilterRoles +from lib389.idm.role import ManagedRoles, FilteredRoles pytestmark = pytest.mark.tier1 @@ -496,7 +496,7 @@ def _create_test_entries(topo): 'cn': 'new managed role'}) # Creating filter role - filters = FilterRoles(topo.standalone, DEFAULT_SUFFIX) + filters = FilteredRoles(topo.standalone, DEFAULT_SUFFIX) filters.create(properties={ 'nsRoleFilter': '(uid=*wal*)', 'description': 'this is the new filtered role', diff --git a/dirsrvtests/tests/suites/roles/basic_test.py b/dirsrvtests/tests/suites/roles/basic_test.py index a3e3c5c..08bc8f2 100644 --- a/dirsrvtests/tests/suites/roles/basic_test.py +++ b/dirsrvtests/tests/suites/roles/basic_test.py @@ -19,7 +19,7 @@ from lib389.idm.user import UserAccount, UserAccounts from lib389.idm.organization import Organization from lib389.idm.organizationalunit import OrganizationalUnit from lib389.topologies import topology_st as topo -from lib389.idm.role import FilterRoles, ManagedRoles, NestedRoles +from lib389.idm.role import FilteredRoles, ManagedRoles, NestedRoles from lib389.idm.domain import Domain pytestmark = pytest.mark.tier1 @@ -59,7 +59,7 @@ def test_filterrole(topo): ou_ou = OrganizationalUnit(topo.standalone, "ou=sales,o=acivattr,{}".format(DEFAULT_SUFFIX)) ou_ou.create(properties=properties) - roles = FilterRoles(topo.standalone, DNBASE) + roles = FilteredRoles(topo.standalone, DNBASE) roles.create(properties={'cn': 'FILTERROLEENGROLE', 'nsRoleFilter': 'cn=eng*'}) roles.create(properties={'cn': 'FILTERROLESALESROLE', 'nsRoleFilter': 'cn=sales*'}) diff --git a/src/lib389/cli/dsidm b/src/lib389/cli/dsidm index 9e5b998..aab6f6f 100755 --- a/src/lib389/cli/dsidm +++ b/src/lib389/cli/dsidm @@ -11,11 +11,10 @@ # PYTHON_ARGCOMPLETE_OK import ldap -import argparse, argcomplete -import logging +import argparse +import argcomplete import sys import signal -from lib389._constants import DN_DM from lib389.cli_idm import account as cli_account from lib389.cli_idm import initialise as cli_init from lib389.cli_idm import organizationalunit as cli_ou @@ -23,6 +22,7 @@ from lib389.cli_idm import group as cli_group from lib389.cli_idm import posixgroup as cli_posixgroup from lib389.cli_idm import user as cli_user from lib389.cli_idm import client_config as cli_client_config +from lib389.cli_idm import role as cli_role from lib389.cli_base import connect_instance, disconnect_instance, setup_script_logger from lib389.cli_base.dsrc import dsrc_to_ldap, dsrc_arg_concat @@ -73,6 +73,7 @@ cli_ou.create_parser(subparsers) cli_posixgroup.create_parser(subparsers) cli_user.create_parser(subparsers) cli_client_config.create_parser(subparsers) +cli_role.create_parser(subparsers) argcomplete.autocomplete(parser) diff --git a/src/lib389/lib389/_mapped_object.py b/src/lib389/lib389/_mapped_object.py index 3953a67..5cfb59e 100644 --- a/src/lib389/lib389/_mapped_object.py +++ b/src/lib389/lib389/_mapped_object.py @@ -1138,14 +1138,19 @@ class DSLdapObjects(DSLogging): # Now actually commit the creation req return co.ensure_state(rdn, properties, self._basedn) - def filter(self, search): + def filter(self, search, scope=None): # This will yield and & filter for objectClass with as many terms as needed. - search_filter = _gen_and([self._get_objectclass_filter(), search]) - self._log.debug('list filter = %s' % search_filter) + if search: + search_filter = _gen_and([self._get_objectclass_filter(), search]) + else: + search_filter = self._get_objectclass_filter() + if scope is None: + scope = self._scope + self._log.debug(f'list filter = {search_filter} with scope {scope}') try: results = self._instance.search_ext_s( base=self._basedn, - scope=self._scope, + scope=scope, filterstr=search_filter, attrlist=self._list_attrlist, serverctrls=self._server_controls, clientctrls=self._client_controls diff --git a/src/lib389/lib389/backend.py b/src/lib389/lib389/backend.py index 5405e20..d2249a4 100644 --- a/src/lib389/lib389/backend.py +++ b/src/lib389/lib389/backend.py @@ -874,4 +874,3 @@ class DatabaseConfig(DSLdapObject): self._create_objectclasses = ['top', 'extensibleObject'] self._protected = True self._dn = "cn=config,cn=ldbm database,cn=plugins,cn=config" - diff --git a/src/lib389/lib389/cli_base/__init__.py b/src/lib389/lib389/cli_base/__init__.py index d139040..932938b 100644 --- a/src/lib389/lib389/cli_base/__init__.py +++ b/src/lib389/lib389/cli_base/__init__.py @@ -11,11 +11,12 @@ import logging import sys import json import ldap +from ldap.dn import is_dn from getpass import getpass from lib389 import DirSrv from lib389.utils import assert_c, get_ldapurl_from_serverid -from lib389.properties import * +from lib389.properties import SER_ROOT_PW, SER_ROOT_DN def _get_arg(args, msg=None, hidden=False, confirm=False): @@ -37,6 +38,13 @@ def _get_arg(args, msg=None, hidden=False, confirm=False): return input("%s : " % msg) +def _get_dn_arg(args, msg=None): + dn_arg = _get_arg(args, msg) + if not is_dn(dn_arg): + raise ValueError(f"{dn_arg} is not a valid DN") + return dn_arg + + def _get_args(args, kws): kwargs = {} while len(kws) > 0: diff --git a/src/lib389/lib389/cli_idm/__init__.py b/src/lib389/lib389/cli_idm/__init__.py index fcf0175..28648ad 100644 --- a/src/lib389/lib389/cli_idm/__init__.py +++ b/src/lib389/lib389/cli_idm/__init__.py @@ -7,8 +7,6 @@ # --- END COPYRIGHT BLOCK --- from getpass import getpass -from lib389 import DirSrv -from lib389.properties import * import json diff --git a/src/lib389/lib389/cli_idm/account.py b/src/lib389/lib389/cli_idm/account.py index f95c6c6..025ea1e 100644 --- a/src/lib389/lib389/cli_idm/account.py +++ b/src/lib389/lib389/cli_idm/account.py @@ -7,9 +7,13 @@ # See LICENSE for details. # --- END COPYRIGHT BLOCK --- +import ldap +import math +import time +from datetime import datetime import argparse -from lib389.idm.account import Account, Accounts +from lib389.idm.account import Account, Accounts, AccountState from lib389.cli_base import ( _generic_get, _generic_get_dn, @@ -17,66 +21,115 @@ from lib389.cli_base import ( _generic_delete, _generic_modify_dn, _get_arg, + _get_dn_arg, _warn, ) +from lib389.utils import gentime_to_posix_time + MANY = Accounts SINGULAR = Account + def list(inst, basedn, log, args): _generic_list(inst, basedn, log.getChild('_generic_list'), MANY, args) + def get_dn(inst, basedn, log, args): - dn = _get_arg( args.dn, msg="Enter dn to retrieve") + dn = _get_dn_arg(args.dn, msg="Enter dn to retrieve") _generic_get_dn(inst, basedn, log.getChild('_generic_get_dn'), MANY, dn, args) + def delete(inst, basedn, log, args, warn=True): - dn = _get_arg( args.dn, msg="Enter dn to delete") + dn = _get_dn_arg(args.dn, msg="Enter dn to delete") if warn: _warn(dn, msg="Deleting %s %s" % (SINGULAR.__name__, dn)) _generic_delete(inst, basedn, log.getChild('_generic_delete'), SINGULAR, dn, args) + def modify(inst, basedn, log, args, warn=True): - dn = _get_arg( args.dn, msg="Enter dn to modify") + dn = _get_dn_arg(args.dn, msg="Enter dn to modify") _generic_modify_dn(inst, basedn, log.getChild('_generic_modify'), MANY, dn, args) -def status(inst, basedn, log, args): - dn = _get_arg( args.dn, msg="Enter dn to check") + +def _print_entry_status(status, dn, log): + log.info(f'Entry DN: {dn}') + for name, value in status["params"].items(): + if "Time" in name and value is not None: + inactivation_date = datetime.fromtimestamp(status["calc_time"] + value) + log.info(f"Entry {name}: {int(math.fabs(value))} seconds ({inactivation_date.strftime('%Y-%m-%d %H:%M:%S')})") + elif "Date" in name and value is not None: + log.info(f"Entry {name}: {value.strftime('%Y%m%d%H%M%SZ')} ({value.strftime('%Y-%m-%d %H:%M:%S')})") + log.info(f'Entry State: {status["state"].describe(status["role_dn"])}\n') + + +def entry_status(inst, basedn, log, args): + dn = _get_dn_arg(args.dn, msg="Enter dn to check") accounts = Accounts(inst, basedn) acct = accounts.get(dn=dn) - acct_str = "locked: %s" % acct.is_locked() - log.info('dn: %s' % dn) - log.info(acct_str) + status = acct.status() + _print_entry_status(status, dn, log) + + +def subtree_status(inst, basedn, log, args): + basedn = _get_dn_arg(args.basedn, msg="Enter basedn to check") + filter = "" + scope = ldap.SCOPE_SUBTREE + epoch_inactive_time = None + if args.scope == "one": + scope = ldap.SCOPE_ONELEVEL + if args.filter: + filter = args.filter + if args.become_inactive_on: + datetime_inactive_time = datetime.strptime(args.become_inactive_on, '%Y-%m-%dT%H:%M:%S') + epoch_inactive_time = datetime.timestamp(datetime_inactive_time) + + account_list = Accounts(inst, basedn).filter(filter, scope) + if not account_list: + raise ValueError(f"No entries were found under {basedn}") + + for entry in account_list: + status = entry.status() + state = status["state"] + params = status["params"] + if args.inactive_only and state == AccountState.ACTIVATED: + continue + if args.become_inactive_on: + if epoch_inactive_time is None or params["Time Until Inactive"] is None or \ + epoch_inactive_time <= (params["Time Until Inactive"] + status["calc_time"]): + continue + _print_entry_status(status, entry.dn, log) + def lock(inst, basedn, log, args): - dn = _get_arg( args.dn, msg="Enter dn to check") + dn = _get_dn_arg(args.dn, msg="Enter dn to check") accounts = Accounts(inst, basedn) acct = accounts.get(dn=dn) acct.lock() - log.info('locked %s' % dn) + log.info(f'Entry {dn} is locked') + def unlock(inst, basedn, log, args): - dn = _get_arg( args.dn, msg="Enter dn to check") + dn = _get_dn_arg(args.dn, msg="Enter dn to check") accounts = Accounts(inst, basedn) acct = accounts.get(dn=dn) acct.unlock() - log.info('unlocked %s' % dn) + log.info(f'Entry {dn} is unlocked') + def reset_password(inst, basedn, log, args): - dn = _get_arg(args.dn, msg="Enter dn to reset password") - new_password = _get_arg(args.new_password, hidden=True, confirm=True, - msg="Enter new password for %s" % dn) + dn = _get_dn_arg(args.dn, msg="Enter dn to reset password") + new_password = _get_arg(args.new_password, hidden=True, confirm=True, msg="Enter new password for %s" % dn) accounts = Accounts(inst, basedn) acct = accounts.get(dn=dn) acct.reset_password(new_password) log.info('reset password for %s' % dn) + def change_password(inst, basedn, log, args): - dn = _get_arg(args.dn, msg="Enter dn to change password") - cur_password = _get_arg(args.current_password, hidden=True, confirm=False, - msg="Enter current password for %s" % dn) - new_password = _get_arg(args.new_password, hidden=True, confirm=True, - msg="Enter new password for %s" % dn) + dn = _get_dn_arg(args.dn, msg="Enter dn to change password") + cur_password = _get_arg(args.current_password, hidden=True, confirm=False, msg="Enter current password for %s" % dn) + new_password = _get_arg(args.new_password, hidden=True, confirm=True, msg="Enter new password for %s" % dn) accounts = Accounts(inst, basedn) acct = accounts.get(dn=dn) acct.change_password(cur_password, new_password) @@ -109,14 +162,25 @@ like modify, locking and unlocking. To create an account, see "user" subcommand lock_parser.set_defaults(func=lock) lock_parser.add_argument('dn', nargs='?', help='The dn to lock') - status_parser = subcommands.add_parser('status', help='status') - status_parser.set_defaults(func=status) - status_parser.add_argument('dn', nargs='?', help='The dn to check') - unlock_parser = subcommands.add_parser('unlock', help='unlock') unlock_parser.set_defaults(func=unlock) unlock_parser.add_argument('dn', nargs='?', help='The dn to unlock') + status_parser = subcommands.add_parser('entry-status', help='status of a single entry') + status_parser.set_defaults(func=entry_status) + status_parser.add_argument('dn', nargs='?', help='The single entry dn to check') + status_parser.add_argument('-V', '--details', action='store_true', help="Print more account policy details about the entry") + + status_parser = subcommands.add_parser('subtree-status', help='status of a subtree') + status_parser.set_defaults(func=subtree_status) + status_parser.add_argument('basedn', help="Search base for finding entries") + status_parser.add_argument('-V', '--details', action='store_true', help="Print more account policy details about the entries") + status_parser.add_argument('-f', '--filter', help="Search filter for finding entries") + status_parser.add_argument('-s', '--scope', choices=['one', 'sub'], help="Search scope (one, sub - default is sub") + status_parser.add_argument('-i', '--inactive-only', action='store_true', help="Only display inactivated entries") + status_parser.add_argument('-o', '--become-inactive-on', + help="Only display entries that will become inactive before specified date (in a format 2007-04-25T14:30)") + reset_pw_parser = subcommands.add_parser('reset_password', help='Reset the password of an account. This should be performed by a directory admin.') reset_pw_parser.set_defaults(func=reset_password) reset_pw_parser.add_argument('dn', nargs='?', help='The dn to reset the password for') @@ -127,5 +191,3 @@ like modify, locking and unlocking. To create an account, see "user" subcommand change_pw_parser.add_argument('dn', nargs='?', help='The dn to change the password for') change_pw_parser.add_argument('new_password', nargs='?', help='The new password to set') change_pw_parser.add_argument('current_password', nargs='?', help='The accounts current password') - - diff --git a/src/lib389/lib389/cli_idm/role.py b/src/lib389/lib389/cli_idm/role.py new file mode 100644 index 0000000..922dadc --- /dev/null +++ b/src/lib389/lib389/cli_idm/role.py @@ -0,0 +1,126 @@ +# --- BEGIN COPYRIGHT BLOCK --- +# Copyright (C) 2019, Red Hat inc, +# Copyright (C) 2018, William Brown +# All rights reserved. +# +# License: GPL (version 3 or any later version). +# See LICENSE for details. +# --- END COPYRIGHT BLOCK --- + +import ldap +import argparse + +from lib389.idm.role import Role, Roles, RoleState +from lib389.cli_base import ( + _generic_get, + _generic_get_dn, + _generic_list, + _generic_delete, + _generic_modify_dn, + _get_arg, + _get_dn_arg, + _warn, + ) + +MANY = Roles +SINGULAR = Role + + +def list(inst, basedn, log, args): + _generic_list(inst, basedn, log.getChild('_generic_list'), MANY, args) + + +def get_dn(inst, basedn, log, args): + dn = _get_dn_arg(args.dn, msg="Enter dn to retrieve") + _generic_get_dn(inst, basedn, log.getChild('_generic_get_dn'), MANY, dn, args) + + +def delete(inst, basedn, log, args, warn=True): + dn = _get_dn_arg(args.dn, msg="Enter dn to delete") + if warn: + _warn(dn, msg="Deleting %s %s" % (SINGULAR.__name__, dn)) + _generic_delete(inst, basedn, log.getChild('_generic_delete'), SINGULAR, dn, args) + + +def modify(inst, basedn, log, args, warn=True): + dn = _get_dn_arg(args.dn, msg="Enter dn to modify") + _generic_modify_dn(inst, basedn, log.getChild('_generic_modify'), MANY, dn, args) + + +def entry_status(inst, basedn, log, args): + dn = _get_dn_arg(args.dn, msg="Enter dn to check") + roles = Roles(inst, basedn) + role = roles.get(dn=dn) + status = role.status() + log.info(f'Entry DN: {dn}') + log.info(f'Entry State: {status["state"].describe(status["role_dn"])}\n') + + +def subtree_status(inst, basedn, log, args): + basedn = _get_dn_arg(args.basedn, msg="Enter basedn to check") + filter = "" + scope = ldap.SCOPE_SUBTREE + + role_list = Roles(inst, basedn).filter(filter, scope) + if not role_list: + raise ValueError(f"No entries were found under {basedn} or the user doesn't have an access") + + for entry in role_list: + status = entry.status() + log.info(f'Entry DN: {entry.dn}') + log.info(f'Entry State: {status["state"].describe(status["role_dn"])}\n') + + +def lock(inst, basedn, log, args): + dn = _get_dn_arg(args.dn, msg="Enter dn to check") + role = Role(inst, dn=dn) + role.lock() + log.info(f'Entry {dn} is locked') + + +def unlock(inst, basedn, log, args): + dn = _get_dn_arg(args.dn, msg="Enter dn to check") + role = Role(inst, dn=dn) + role.unlock() + log.info(f'Entry {dn} is unlocked') + + +def create_parser(subparsers): + role_parser = subparsers.add_parser('role', help='''Manage generic roles, with tasks +like modify, locking and unlocking.''') + + subcommands = role_parser.add_subparsers(help='action') + + list_parser = subcommands.add_parser('list', help='list roles that could login to the directory') + list_parser.set_defaults(func=list) + + get_dn_parser = subcommands.add_parser('get-by-dn', help='get-by-dn ') + get_dn_parser.set_defaults(func=get_dn) + get_dn_parser.add_argument('dn', nargs='?', help='The dn to get and display') + + modify_dn_parser = subcommands.add_parser('modify-by-dn', help='modify-by-dn :: ...') + modify_dn_parser.set_defaults(func=modify) + modify_dn_parser.add_argument('dn', nargs=1, help='The dn to get and display') + modify_dn_parser.add_argument('changes', nargs='+', help="A list of changes to apply in format: ::") + + delete_parser = subcommands.add_parser('delete', help='deletes the role') + delete_parser.set_defaults(func=delete) + delete_parser.add_argument('dn', nargs='?', help='The dn of the role to delete') + + lock_parser = subcommands.add_parser('lock', help='lock') + lock_parser.set_defaults(func=lock) + lock_parser.add_argument('dn', nargs='?', help='The dn to lock') + + unlock_parser = subcommands.add_parser('unlock', help='unlock') + unlock_parser.set_defaults(func=unlock) + unlock_parser.add_argument('dn', nargs='?', help='The dn to unlock') + + status_parser = subcommands.add_parser('entry-status', help='status of a single entry') + status_parser.set_defaults(func=entry_status) + status_parser.add_argument('dn', nargs='?', help='The single entry dn to check') + + status_parser = subcommands.add_parser('subtree-status', help='status of a subtree') + status_parser.set_defaults(func=subtree_status) + status_parser.add_argument('basedn', help="Search base for finding entries") + status_parser.add_argument('-f', '--filter', help="Search filter for finding entries") + status_parser.add_argument('-s', '--scope', choices=['base', 'one', 'sub'], help="Search scope (base, one, sub - default is sub") diff --git a/src/lib389/lib389/idm/account.py b/src/lib389/lib389/idm/account.py index 4a5c5c4..c95e7a9 100644 --- a/src/lib389/lib389/idm/account.py +++ b/src/lib389/lib389/idm/account.py @@ -1,4 +1,5 @@ # --- BEGIN COPYRIGHT BLOCK --- +# Copyright (C) 2019 Red Hat, Inc. # Copyright (C) 2017, William Brown # All rights reserved. # @@ -6,12 +7,33 @@ # See LICENSE for details. # --- END COPYRIGHT BLOCK --- +import os +import time +import subprocess +from enum import Enum +import ldap + from lib389._mapped_object import DSLdapObject, DSLdapObjects, _gen_or, _gen_filter, _term_gen from lib389._constants import SER_ROOT_DN, SER_ROOT_PW -from lib389.utils import ds_is_older +from lib389.utils import ds_is_older, gentime_to_posix_time, gentime_to_datetime +from lib389.plugins import AccountPolicyPlugin, AccountPolicyConfig, AccountPolicyEntry +from lib389.cos import CosTemplates +from lib389.mappingTree import MappingTrees +from lib389.idm.role import Roles + + +class AccountState(Enum): + ACTIVATED = "activated" + DIRECTLY_LOCKED = "directly locked through nsAccountLock" + INDIRECTLY_LOCKED = "indirectly locked through a Role" + INACTIVITY_LIMIT_EXCEEDED = "inactivity limit exceeded" + + def describe(self, role_dn=None): + if self.name == "INDIRECTLY_LOCKED" and role_dn is not None: + return f'{self.value} - {role_dn}' + else: + return f'{self.value}' -import os -import subprocess class Account(DSLdapObject): """A single instance of Account entry @@ -22,23 +44,152 @@ class Account(DSLdapObject): :type dn: str """ - def is_locked(self): - """Check if nsAccountLock is set - - :returns: True if account is locked + def _format_status_message(self, message, create_time, modify_time, last_login_time, limit, role_dn=None): + params = {} + now = time.time() + params["Creation Date"] = gentime_to_datetime(create_time) + params["Modification Date"] = gentime_to_datetime(modify_time) + params["Last Login Date"] = None + params["Time Until Inactive"] = None + params["Time Since Inactive"] = None + if last_login_time: + params["Last Login Date"] = gentime_to_datetime(last_login_time) + if limit: + remaining_time = float(limit) + gentime_to_posix_time(last_login_time) - now + if remaining_time <= 0: + if message == AccountState.INACTIVITY_LIMIT_EXCEEDED: + params["Time Since Inactive"] = remaining_time + else: + params["Time Until Inactive"] = remaining_time + result = {"state": message, "params": params, "calc_time": now, "role_dn": None} + if role_dn is not None: + result["role_dn"] = role_dn + return result + + def _dict_get_with_ignore_indexerror(self, dict, attr): + try: + return dict[attr][0] + except IndexError: + return "" + + def status(self): + """Check if account is locked by Account Policy plugin or + nsAccountLock (directly or indirectly) + + :returns: a dict in a format - + {"status": status, "params": activity_data, "calc_time": epoch_time} """ - return self.present('nsAccountLock') + inst = self._instance + + # Fetch Account Policy data if its enabled + plugin = AccountPolicyPlugin(inst) + state_attr = "" + alt_state_attr = "" + limit = "" + spec_attr = "" + limit_attr = "" + process_account_policy = False + try: + process_account_policy = plugin.status() + except IndexError: + self._log.debug("The bound user doesn't have rights to access Account Policy settings. Not checking.") + + if process_account_policy: + config_dn = plugin.get_attr_val_utf8("nsslapd-pluginarg0") + config = AccountPolicyConfig(inst, config_dn) + config_settings = config.get_attrs_vals_utf8(["stateattrname", "altstateattrname", + "specattrname", "limitattrname"]) + state_attr = self._dict_get_with_ignore_indexerror(config_settings, "stateattrname") + alt_state_attr = self._dict_get_with_ignore_indexerror(config_settings, "altstateattrname") + spec_attr = self._dict_get_with_ignore_indexerror(config_settings, "specattrname") + limit_attr = self._dict_get_with_ignore_indexerror(config_settings, "limitattrname") + + cos_entries = CosTemplates(inst, self.dn) + accpol_entry_dn = "" + for cos in cos_entries.list(): + if cos.present(spec_attr): + accpol_entry_dn = cos.get_attr_val_utf8_l(spec_attr) + if accpol_entry_dn: + accpol_entry = AccountPolicyEntry(inst, accpol_entry_dn) + else: + accpol_entry = config + limit = accpol_entry.get_attr_val_utf8_l(limit_attr) + + # Fetch account data + account_data = self.get_attrs_vals_utf8(["createTimestamp", "modifyTimeStamp", + "nsAccountLock", state_attr]) + + last_login_time = self._dict_get_with_ignore_indexerror(account_data, state_attr) + if not last_login_time: + last_login_time = self._dict_get_with_ignore_indexerror(account_data, alt_state_attr) + + create_time = self._dict_get_with_ignore_indexerror(account_data, "createTimestamp") + modify_time = self._dict_get_with_ignore_indexerror(account_data, "modifyTimeStamp") + + acct_roles = self.get_attr_vals_utf8_l("nsRole") + mapping_trees = MappingTrees(inst) + root_suffix = "" + try: + root_suffix = mapping_trees.get_root_suffix_by_entry(self.dn) + except ldap.NO_SUCH_OBJECT: + self._log.debug("The bound user doesn't have rights to access disabled roles settings. Not checking.") + if root_suffix: + roles = Roles(inst, root_suffix) + try: + disabled_roles = roles.get_disabled_roles() + + # Locked indirectly through a role + locked_indirectly_role_dn = "" + for role in acct_roles: + if str.lower(role) in [str.lower(role.dn) for role in disabled_roles.keys()]: + locked_indirectly_role_dn = role + if locked_indirectly_role_dn: + return self._format_status_message(AccountState.INDIRECTLY_LOCKED, create_time, modify_time, + last_login_time, limit, locked_indirectly_role_dn) + except ldap.NO_SUCH_OBJECT: + pass + + # Locked directly + if self._dict_get_with_ignore_indexerror(account_data, "nsAccountLock") == "true": + return self._format_status_message(AccountState.DIRECTLY_LOCKED, + create_time, modify_time, last_login_time, limit) + + # Locked indirectly through Account Policy plugin + if process_account_policy and last_login_time: + # Now check the Acount Policy Plugin inactivity limits + remaining_time = float(limit) - (time.time() - gentime_to_posix_time(last_login_time)) + if remaining_time <= 0: + return self._format_status_message(AccountState.INACTIVITY_LIMIT_EXCEEDED, + create_time, modify_time, last_login_time, limit) + # All checks are passed - we are active + return self._format_status_message(AccountState.ACTIVATED, create_time, modify_time, last_login_time, limit) + + def ensure_lock(self): + """Ensure nsAccountLock is set to 'true'""" + + self.replace('nsAccountLock', 'true') + + def ensure_unlock(self): + """Unset nsAccountLock if it's set""" + + self.ensure_removed('nsAccountLock', None) def lock(self): """Set nsAccountLock to 'true'""" + current_status = self.status() + if current_status["state"] == AccountState.DIRECTLY_LOCKED: + raise ValueError("Account is already active") self.replace('nsAccountLock', 'true') def unlock(self): """Unset nsAccountLock""" - self.ensure_removed('nsAccountLock', None) + current_status = self.status() + if current_status["state"] == AccountState.ACTIVATED: + raise ValueError("Account is already active") + self.remove('nsAccountLock', None) # If the account can be bound to, this will attempt to do so. We don't check # for exceptions, just pass them back! @@ -143,6 +294,7 @@ class Account(DSLdapObject): self._instance.passwd_s(self._dn, current_password, new_password, serverctrls=self._server_controls, clientctrls=self._client_controls, escapehatch='i am sure') + class Accounts(DSLdapObjects): """DSLdapObjects that represents Account entry diff --git a/src/lib389/lib389/idm/role.py b/src/lib389/lib389/idm/role.py index b630bff..fe91aab 100644 --- a/src/lib389/lib389/idm/role.py +++ b/src/lib389/lib389/idm/role.py @@ -7,157 +7,334 @@ # --- END COPYRIGHT BLOCK ---- +from enum import Enum +import ldap from lib389._mapped_object import DSLdapObject, DSLdapObjects +from lib389.cos import CosTemplates, CosClassicDefinitions +from lib389.mappingTree import MappingTrees +from lib389.idm.nscontainer import nsContainers -class FilterRole(DSLdapObject): - """A single instance of FilterRole entry to create FilterRole role. +class RoleState(Enum): + ACTIVATED = "activated" + DIRECTLY_LOCKED = "directly locked through nsDisabledRole" + INDIRECTLY_LOCKED = "indirectly locked through a Role" + PROBABLY_ACTIVATED = '''probably activated or nsDisabledRole setup and its CoS entries are not +in a valid state or there is no access to the settings.''' - :param instance: An instance - :type instance: lib389.DirSrv - :param dn: Entry DN - :type dn: str - Usages: - user1 = 'cn=anuj,ou=people,dc=example,ed=com' - user2 = 'cn=unknownuser,ou=people,dc=example,ed=com' - role=FilterRole(topo.standalone,'cn=NameofRole,ou=People,dc=example,dc=com') - role_props={'cn':'Anuj', 'nsRoleFilter':'cn=anuj*'} - role.create(properties=role_props, basedn=SUFFIX) - The user1 entry matches the filter (possesses the cn=anuj* attribute with the value anuj) - therefore, it is a member of this filtered role automatically. + def describe(self, role_dn=None): + if self.name == "INDIRECTLY_LOCKED" and role_dn is not None: + return f'{self.value} - {role_dn}' + else: + return f'{self.value}' + + +class Role(DSLdapObject): + """A single instance of Role entry + + :param instance: An instance + :type instance: lib389.DirSrv + :param dn: Entry DN + :type dn: str """ + def __init__(self, instance, dn=None): - super(FilterRole, self).__init__(instance, dn) + super(Role, self).__init__(instance, dn) self._rdn_attribute = 'cn' self._create_objectclasses = [ 'top', + 'LDAPsubentry', 'nsRoleDefinition', - 'nsComplexRoleDefinition', - 'nsFilteredRoleDefinition' ] + def _format_status_message(self, message, role_dn=None): + return {"state": message, "role_dn": role_dn} + + def status(self): + """Check if role is locked in nsDisabledRole (directly or indirectly) -class FilterRoles(DSLdapObjects): - """DSLdapObjects that represents all filtertrole entries in suffix. - - This instance is used mainly for search operation filtred role - - :param instance: An instance - :type instance: lib389.DirSrv - :param basedn: Suffix DN - :type basedn: str - :param rdn: The DN that will be combined wit basedn - :type rdn: str - Usages: - role_props={'cn':'Anuj', 'nsRoleFilter':'cn=*'} - FilterRoles(topo.standalone, DEFAULT_SUFFIX).create(properties=role_props) - FilterRoles(topo.standalone, DEFAULT_SUFFIX).list() - user1 = 'cn=anuj,ou=people,dc=example,ed=com' - user2 = 'uid=unknownuser,ou=people,dc=example,ed=com' - The user1 entry matches the filter (possesses the cn=* attribute with the value cn) - therefore, it is a member of this filtered role automatically. + :returns: a dict """ + + inst = self._instance + disabled_roles = {} + try: + mapping_trees = MappingTrees(inst) + root_suffix = mapping_trees.get_root_suffix_by_entry(self.dn) + roles = Roles(inst, root_suffix) + disabled_roles = roles.get_disabled_roles() + nested_roles = NestedRoles(inst, root_suffix) + disabled_role = nested_roles.get("nsDisabledRole") + inact_containers = nsContainers(inst, basedn=root_suffix) + inact_container = inact_containers.get('nsAccountInactivationTmp') + + cos_templates = CosTemplates(inst, inact_container.dn) + cos_template = cos_templates.get(f'{disabled_role.dn}') + cos_template.present('cosPriority', '1') + cos_template.present('nsAccountLock', 'true') + + cos_classic_defs = CosClassicDefinitions(inst, root_suffix) + cos_classic_def = cos_classic_defs.get('nsAccountInactivation_cos') + cos_classic_def.present('cosAttribute', 'nsAccountLock operational') + cos_classic_def.present('cosTemplateDn', inact_container.dn) + cos_classic_def.present('cosSpecifier', 'nsRole') + except ldap.NO_SUCH_OBJECT: + return self._format_status_message(RoleState.PROBABLY_ACTIVATED) + + for role, parent in disabled_roles.items(): + if str.lower(self.dn) == str.lower(role.dn): + if parent is None: + return self._format_status_message(RoleState.DIRECTLY_LOCKED) + else: + return self._format_status_message(RoleState.INDIRECTLY_LOCKED, parent) + + return self._format_status_message(RoleState.ACTIVATED) + + def lock(self): + """Set the entry dn to nsDisabledRole and ensure it exists""" + + current_status = self.status() + if current_status["state"] == RoleState.DIRECTLY_LOCKED: + raise ValueError(f"Role is already {current_status['state'].describe()}") + + inst = self._instance + + mapping_trees = MappingTrees(inst) + root_suffix = "" + root_suffix = mapping_trees.get_root_suffix_by_entry(self.dn) + + if root_suffix: + managed_roles = ManagedRoles(inst, root_suffix) + managed_role = managed_roles.ensure_state(properties={"cn": "nsManagedDisabledRole"}) + nested_roles = NestedRoles(inst, root_suffix) + try: + disabled_role = nested_roles.get("nsDisabledRole") + except ldap.NO_SUCH_OBJECT: + # We don't use "ensure_state" because we want to preserve the existing attributes + disabled_role = nested_roles.create(properties={"cn": "nsDisabledRole", + "nsRoleDN": managed_role.dn}) + disabled_role.add("nsRoleDN", self.dn) + + inact_containers = nsContainers(inst, basedn=root_suffix) + inact_container = inact_containers.ensure_state(properties={'cn': 'nsAccountInactivationTmp'}) + + cos_templates = CosTemplates(inst, inact_container.dn) + cos_templates.ensure_state(properties={'cosPriority': '1', + 'nsAccountLock': 'true', + 'cn': f'{disabled_role.dn}'}) + + cos_classic_defs = CosClassicDefinitions(inst, root_suffix) + cos_classic_defs.ensure_state(properties={'cosAttribute': 'nsAccountLock operational', + 'cosSpecifier': 'nsRole', + 'cosTemplateDn': inact_container.dn, + 'cn': 'nsAccountInactivation_cos'}) + + def unlock(self): + """Remove the entry dn from nsDisabledRole if it exists""" + + inst = self._instance + current_status = self.status() + if current_status["state"] == RoleState.ACTIVATED: + raise ValueError("Role is already active") + + mapping_trees = MappingTrees(inst) + root_suffix = mapping_trees.get_root_suffix_by_entry(self.dn) + roles = NestedRoles(inst, root_suffix) + try: + disabled_role = roles.get("nsDisabledRole") + # Still we want to ensure that it is not locked directly too + disabled_role.ensure_removed("nsRoleDN", self.dn) + except ldap.NO_SUCH_OBJECT: + pass + + # Notify if it's locked indirectly + if current_status["state"] == RoleState.INDIRECTLY_LOCKED: + raise ValueError(f"Role is {current_status['state'].describe(current_status['role_dn'])}. Please, deal with it separately") + + +class Roles(DSLdapObjects): + """DSLdapObjects that represents all Roles entries + + :param instance: An instance + :type instance: lib389.DirSrv + :param basedn: Suffix DN + :type basedn: str + """ + def __init__(self, instance, basedn): - super(FilterRoles, self).__init__(instance) + super(Roles, self).__init__(instance) self._objectclasses = [ 'top', + 'LDAPsubentry', 'nsRoleDefinition', - 'nsComplexRoleDefinition', - 'nsFilteredRoleDefinition' ] self._filterattrs = ['cn'] self._basedn = basedn - self._childobject = FilterRole - + self._childobject = Role -class ManagedRole(DSLdapObject): - """A single instance of ManagedRole entry to create ManagedRole role. + def get_with_type(self, selector=[], dn=None): + """Get the correct role type - :param instance: An instance - :type instance: lib389.DirSrv - :param dn: Entry DN + :param dn: DN of wanted entry :type dn: str + :param selector: An additional filter to search for, i.e. 'backend_name'. The attributes + selected are based on object type, ie user will search for uid and cn. + :type dn: str + + :returns: FilteredRole, ManagedRole or NestedRole + """ + + ROLE_OBJECTCLASSES = {FilteredRole: ['nscomplexroledefinition', + 'nsfilteredroledefinition'], + ManagedRole: ['nssimpleroledefinition', + 'nsmanagedroledefinition'], + NestedRole: ['nscomplexroledefinition', + 'nsnestedroledefinition']} + entry = self.get(selector=selector, dn=dn, json=False) + entry_objectclasses = entry.get_attr_vals_utf8_l("objectClass") + role_found = False + for role, objectclasses in ROLE_OBJECTCLASSES.items(): + role_found = all(oc in entry_objectclasses for oc in objectclasses) + if role_found: + return role(self._instance, entry.dn) + if not role_found: + raise ldap.NO_SUCH_OBJECT("Role definition was not found") + + def get_disabled_roles(self): + """Get disabled roles that are usually defined in the cn=nsDisabledRole,ROOT_SUFFIX + + :returns: A dict {role: its_parent, } + """ + + disabled_role = self.get("nsDisabledRole") + roles_inactive = {} + result = {} + + # Do this on 0 level of nestedness + for role_dn in disabled_role.get_attr_vals_utf8_l("nsRoleDN"): + roles_inactive[role_dn] = None + + # We go through the list and check if the role is Nested and + # then add its 'nsrole' attributes to the processing list + while roles_inactive.items(): + processing_role_dn, parent = roles_inactive.popitem() + # Check if already seen the role and skip it then + if processing_role_dn in result.keys(): + continue + + processing_role = self.get_with_type(dn=processing_role_dn) + if isinstance(processing_role, NestedRole): + for role_dn in processing_role.get_attr_vals_utf8_l("nsRoleDN"): + # We don't need to process children which are already present in the list + if role_dn in result.keys() or role_dn in roles_inactive.keys(): + continue + # We are deeper - return its children to the processing and assign the original parent + if parent in [role.dn for role in result.keys()]: + roles_inactive[role_dn] = parent + else: + roles_inactive[role_dn] = processing_role_dn + # Set the processed role to list + result[processing_role] = parent + + return result +class FilteredRole(Role): + """A single instance of FilteredRole entry to create FilteredRole role + + :param instance: An instance + :type instance: lib389.DirSrv + :param dn: Entry DN + :type dn: str """ + + def __init__(self, instance, dn=None): + super(FilteredRole, self).__init__(instance, dn) + self._rdn_attribute = 'cn' + self._create_objectclasses = ['nsComplexRoleDefinition', 'nsFilteredRoleDefinition'] + + + +class FilteredRoles(Roles): + """DSLdapObjects that represents all filtered role entries + + :param instance: An instance + :type instance: lib389.DirSrv + :param basedn: Suffix DN + :type basedn: str + """ + + def __init__(self, instance, basedn): + super(FilteredRoles, self).__init__(instance, basedn) + self._objectclasses = ['LDAPsubentry', 'nsComplexRoleDefinition', 'nsFilteredRoleDefinition'] + self._filterattrs = ['cn'] + self._basedn = basedn + self._childobject = FilteredRole + + +class ManagedRole(Role): + """A single instance of Managed Role entry + + :param instance: An instance + :type instance: lib389.DirSrv + :param dn: Entry DN + :type dn: str + """ + def __init__(self, instance, dn=None): super(ManagedRole, self).__init__(instance, dn) self._rdn_attribute = 'cn' - self._create_objectclasses = [ - 'top', - 'nsRoleDefinition', - 'nsSimpleRoleDefinition', - 'nsManagedRoleDefinition' - ] + self._create_objectclasses = ['nsSimpleRoleDefinition', 'nsManagedRoleDefinition'] -class ManagedRoles(DSLdapObjects): - """DSLdapObjects that represents all ManagedRoles entries in suffix. +class ManagedRoles(Roles): + """DSLdapObjects that represents all Managed Roles entries - This instance is used mainly for search operation ManagedRoles role + :param instance: An instance + :type instance: lib389.DirSrv + :param basedn: Suffix DN + :type basedn: str + :param rdn: The DN that will be combined wit basedn + :type rdn: str + """ - :param instance: An instance - :type instance: lib389.DirSrv - :param basedn: Suffix DN - :type basedn: str - :param rdn: The DN that will be combined wit basedn - :type rdn: str - """ def __init__(self, instance, basedn): - super(ManagedRoles, self).__init__(instance) - self._objectclasses = [ - 'top', - 'nsRoleDefinition', - 'nsSimpleRoleDefinition', - 'nsManagedRoleDefinition' - ] + super(ManagedRoles, self).__init__(instance, basedn) + self._objectclasses = ['LDAPsubentry', 'nsSimpleRoleDefinition', 'nsManagedRoleDefinition'] self._filterattrs = ['cn'] self._basedn = basedn self._childobject = ManagedRole -class NestedRole(DSLdapObject): - """A single instance of NestedRole entry to create NestedRole role. - - :param instance: An instance - :type instance: lib389.DirSrv - :param dn: Entry DN - :type dn: str +class NestedRole(Role): + """A single instance of Nested Role entry + :param instance: An instance + :type instance: lib389.DirSrv + :param dn: Entry DN + :type dn: str """ + def __init__(self, instance, dn=None): super(NestedRole, self).__init__(instance, dn) self._must_attributes = ['cn', 'nsRoleDN'] self._rdn_attribute = 'cn' - self._create_objectclasses = [ - 'top', - 'nsRoleDefinition', - 'nsComplexRoleDefinition', - 'ldapSubEntry', - 'nsNestedRoleDefinition' - ] + self._create_objectclasses = ['nsComplexRoleDefinition', 'nsNestedRoleDefinition'] -class NestedRoles(DSLdapObjects): +class NestedRoles(Roles): """DSLdapObjects that represents all NestedRoles entries in suffix. - This instance is used mainly for search operation NestedRoles role + :param instance: An instance + :type instance: lib389.DirSrv + :param basedn: Suffix DN + :type basedn: str + :param rdn: The DN that will be combined wit basedn + :type rdn: str + """ - :param instance: An instance - :type instance: lib389.DirSrv - :param basedn: Suffix DN - :type basedn: str - :param rdn: The DN that will be combined wit basedn - :type rdn: str - """ def __init__(self, instance, basedn): - super(NestedRoles, self).__init__(instance) - self._objectclasses = [ - 'top', - 'nsRoleDefinition', - 'nsComplexRoleDefinition', - 'ldapSubEntry', - 'nsNestedRoleDefinition' - ] + super(NestedRoles, self).__init__(instance, basedn) + self._objectclasses = ['LDAPsubentry', 'nsComplexRoleDefinition', 'nsNestedRoleDefinition'] self._filterattrs = ['cn'] self._basedn = basedn self._childobject = NestedRole diff --git a/src/lib389/lib389/mappingTree.py b/src/lib389/lib389/mappingTree.py index 7a59020..45f1a9a 100644 --- a/src/lib389/lib389/mappingTree.py +++ b/src/lib389/lib389/mappingTree.py @@ -7,7 +7,7 @@ # --- END COPYRIGHT BLOCK --- import ldap -import ldap.dn +from ldap.dn import str2dn, dn2str import six from lib389._constants import * @@ -395,7 +395,7 @@ class MappingTree(DSLdapObject): super(MappingTree, self).__init__(instance, dn) self._rdn_attribute = 'cn' self._must_attributes = ['cn'] - self._create_objectclasses = ['top', 'extensibleObject', MT_OBJECTCLASS_VALUE] + self._create_objectclasses = ['top', 'extensibleObject', 'nsMappingTree'] self._protected = False def set_parent(self, parent): @@ -422,9 +422,31 @@ class MappingTrees(DSLdapObjects): def __init__(self, instance): super(MappingTrees, self).__init__(instance=instance) - self._objectclasses = [MT_OBJECTCLASS_VALUE] - self._filterattrs = ['cn', 'nsslapd-backend' ] + self._objectclasses = ['nsMappingTree'] + self._filterattrs = ['cn', 'nsslapd-backend'] self._childobject = MappingTree self._basedn = DN_MAPPING_TREE + def get_root_suffix_by_entry(self, entry_dn): + """Get the root suffix to which the entry belongs + :param entry_dn: An entry DN + :type entry_dn: str + :returns: str + """ + + mapping_tree_list = sorted(self.list(), key=lambda b: len(b.dn), reverse=True) + + entry_dn_parts = str2dn(entry_dn) + processing = True + while processing: + compare_dn = dn2str(entry_dn_parts) + for mapping_tree in mapping_tree_list: + if str.lower(compare_dn) == str.lower(mapping_tree.rdn): + processing = False + return mapping_tree.rdn + if entry_dn_parts: + entry_dn_parts.pop(0) + else: + processing = False + raise ldap.NO_SUCH_OBJECT(f"{entry_dn} doesn't belong to any suffix") diff --git a/src/lib389/lib389/plugins.py b/src/lib389/lib389/plugins.py index 8271e03..a8b13c7 100644 --- a/src/lib389/lib389/plugins.py +++ b/src/lib389/lib389/plugins.py @@ -1820,6 +1820,40 @@ class AccountPolicyConfigs(DSLdapObjects): self._basedn = basedn +class AccountPolicyEntry(DSLdapObject): + """A single instance of Account Policy Plugin entry which is used for CoS + + :param instance: An instance + :type instance: lib389.DirSrv + :param dn: Entry DN + :type dn: str + """ + + def __init__(self, instance, dn=None): + super(AccountPolicyConfig, self).__init__(instance, dn) + self._rdn_attribute = 'cn' + self._must_attributes = ['cn'] + self._create_objectclasses = ['top', 'accountpolicy'] + self._protected = False + + +class AccountPolicyEntries(DSLdapObjects): + """A DSLdapObjects entity which represents Account Policy Plugin entry which is used for CoS + + :param instance: An instance + :type instance: lib389.DirSrv + :param basedn: Base DN for all account entries below + :type basedn: str + """ + + def __init__(self, instance, basedn): + super(AccountPolicyConfigs, self).__init__(instance) + self._objectclasses = ['top', 'accountpolicy'] + self._filterattrs = ['cn'] + self._childobject = AccountPolicyEntry + self._basedn = basedn + + class DNAPlugin(Plugin): """A single instance of Distributed Numeric Assignment plugin entry diff --git a/src/lib389/lib389/utils.py b/src/lib389/lib389/utils.py index 10c8cae..51960eb 100644 --- a/src/lib389/lib389/utils.py +++ b/src/lib389/lib389/utils.py @@ -31,6 +31,7 @@ import shutil import ldap import socket import time +from datetime import datetime import sys import filecmp import pwd @@ -1085,6 +1086,29 @@ def ds_is_newer(*ver): return ds_is_related('newer', *ver) +def gentime_to_datetime(gentime): + """Convert Generalized time to datetime object + + :param gentime: Time in the format - YYYYMMDDHHMMSSZ (20170126120000Z) + :type password: str + :returns: datetime.datetime object + """ + + return datetime.strptime(gentime, '%Y%m%d%H%M%SZ') + + +def gentime_to_posix_time(gentime): + """Convert Generalized time to POSIX time format + + :param gentime: Time in the format - YYYYMMDDHHMMSSZ (20170126120000Z) + :type password: str + :returns: Epoch time int + """ + + target_timestamp = gentime_to_datetime(gentime) + return datetime.timestamp(target_timestamp) + + def getDateTime(): """ Return the date and time: