# Copyright (C) 2025 Red Hat # see file 'COPYING' for use and warranty information from ipalib import api, errors from ipalib import Str, Bool, Password, Flag from ipalib.plugable import Registry from ipalib.request import context from .baseldap import ( pkey_to_value, LDAPObject, LDAPCreate, LDAPDelete, LDAPUpdate, LDAPSearch, LDAPRetrieve, LDAPQuery) from .baseuser import validate_nsaccountlock, convert_nsaccountlock from ipalib import _, ngettext from ipalib import constants from ipalib import output from ipalib.messages import ServerSysacctMgrUpdateRequired, SystemAccountUsage from ipapython.ipautil import ipa_generate_password, TMP_PWD_ENTROPY_BITS from ipapython.dn import DN __doc__ = _(""" System accounts System accounts designed to allow applications to query LDAP database. Unlike IPA users, system accounts have no POSIX properties and cannot be resolved as 'users' in a POSIX environment. System accounts are stored in cn=sysaccounts,cn=etc LDAP subtree. Some of system accounts are special to IPA's own operations and cannot be removed. EXAMPLES: Add a new system account, set random password: ipa sysaccount-add my-app --random Allow the system account to change user passwords without triggering a reset: ipa sysaccount-mod my-app --privileged=True The system account still needs to be permitted to modify user passwords through a role that includes a corresponding permission ('System: Change User password'), through the privilege system: ipa privilege-add 'my-app password change privilege' ipa privilege-add-permission 'my-app password change privilege' \ --permission 'System: Change User password' ipa role-add 'my-app role' ipa role-add-privilege 'my-app role' \ --privilege 'my-app password change privilege' ipa role-add-member 'my-app role' --sysaccounts my-app Delete a system account: ipa sysaccount-del my-app Find all system accounts: ipa sysaccount-find Disable the system account: ipa sysaccount-disable my-app Re-enable the system account: ipa sysaccount-enable my-app Allow the system account to change user passwords without a reset: ipa sysaccount-policy my-app --privileged=true """) register = Registry() required_system_accounts = [ 'passsync', 'sudo', ] sysaccount_mgrs_dn = DN('cn=ipa_pwd_extop,cn=plugins,cn=config') attr_sysacctmgrdns = 'sysacctmanagersdns' update_without_reset = ( Bool( 'privileged?', label=_('Privileged'), doc=_('Allow password updates without reset'), ), ) def check_userpassword(entry_attrs, **options): if 'userpassword' not in entry_attrs and options.get('random'): entry_attrs['userpassword'] = ipa_generate_password( entropy_bits=TMP_PWD_ENTROPY_BITS) # save the password so it can be displayed in post_callback setattr(context, 'randompassword', entry_attrs['userpassword']) def fill_randompassword(entry_attrs, **options): if options.get('random', False): try: entry_attrs['randompassword'] = getattr(context, 'randompassword') except AttributeError: # if both randompassword and userpassword options were used pass @register() class sysaccount(LDAPObject): """ System account object. """ container_dn = api.env.container_sysaccounts object_name = _('system account') object_name_plural = _('system accounts') object_class = [ 'account', 'simplesecurityobject' ] possible_objectclasses = ['nsmemberof'] permission_filter_objectclasses = ['simplesecurityobject'] search_attributes = ['uid', 'description'] default_attributes = [ 'uid', 'description', 'memberof', 'nsaccountlock'] uuid_attribute = '' attribute_members = { 'memberof': ['role'], } password_attributes = [('userpassword', 'has_password')] bindable = True relationships = { 'managedby': ('Managed by', 'man_by_', 'not_man_by_'), } password_attributes = [('userpassword', 'has_password')] managed_permissions = { 'System: Read System Accounts': { 'ipapermbindruletype': 'all', 'ipapermright': {'read', 'search', 'compare'}, 'ipapermdefaultattr': { 'objectclass', 'uid', 'memberof', 'nsaccountlock', 'description' }, }, 'System: Check System Accounts passwords': { 'ipapermright': {'search'}, 'ipapermdefaultattr': {'userpassword'}, 'default_privileges': {'System Accounts Administrators'}, }, 'System: Add System Accounts': { 'ipapermright': {'add'}, 'default_privileges': {'System Accounts Administrators'}, }, 'System: Modify System Accounts': { 'ipapermright': {'write'}, 'ipapermdefaultattr': {'userpassword', 'description', 'nsaccountlock'}, 'default_privileges': {'System Accounts Administrators'}, }, 'System: Remove System Accounts': { 'ipapermright': {'delete'}, 'default_privileges': {'System Accounts Administrators'}, }, } label = _('System Accounts') label_singular = _('System Account') takes_params = ( Str('uid', pattern=constants.PATTERN_GROUPUSER_NAME, pattern_errmsg=constants.ERRMSG_GROUPUSER_NAME.format('user'), maxlength=255, cli_name='login', label=_('System account ID'), primary_key=True, normalizer=lambda value: value.lower()), Str('description?', cli_name='desc', doc=_('A description of system account'), label=_('Description')), Password('userpassword?', cli_name='password', label=_('Password'), doc=_('Prompt to set the user password'), exclude='webui', flags=('no_search',)), Flag('random?', doc=_('Generate a random user password'), flags=('no_search', 'virtual_attribute'), default=False), Str('randompassword?', label=_('Random password'), flags=('no_create', 'no_update', 'no_search', 'virtual_attribute')), Bool('nsaccountlock?', cli_name=('disabled'), default=False, label=_('Account disabled')), ) def get_dn(self, *keys, **kwargs): key = keys[0] parent_dn = DN(self.container_dn, self.api.env.basedn) true_rdn = 'uid' return self.backend.make_dn_from_attr( true_rdn, key, parent_dn ) def get_password_attributes(self, ldap, dn, entry_attrs): """ Search on the entry to determine if it has a password or keytab set. """ # Limit objectclass to simpleSecurityObject obj_filter = self.api.Object.permission.make_type_filter(self) for (pwattr, attr) in self.password_attributes: search_filter = '(&(%s=*)%s)' % (pwattr, obj_filter) try: ldap.find_entries( search_filter, [pwattr], dn, ldap.SCOPE_BASE ) entry_attrs[attr] = True except errors.NotFound: entry_attrs[attr] = False def handle_reset(self, cmd, next_cmd, ldap, dn, entry_attrs, **options): privileged = None exc = None if 'privileged' in options: # TODO: change the code to perform DBUS oddjob operation instead # because cn=config changes require cn=Directory Manager permissions # and then 389-ds needs a restart add_to_passsync_mgrs = options.get('privileged', False) try: if add_to_passsync_mgrs: ldap.add_entry_to_group( dn, sysaccount_mgrs_dn, attr_sysacctmgrdns) privileged = True else: ldap.remove_entry_from_group( dn, sysaccount_mgrs_dn, attr_sysacctmgrdns) privileged = False if next_cmd: command_name = next_cmd.name.replace('_','-') cmd.add_message(ServerSysacctMgrUpdateRequired( server=cmd.api.env.server, command=command_name)) except (errors.EmptyModlist, errors.NotGroupMember, errors.AlreadyGroupMember) as e: exc = e if entry_attrs is not None: if privileged is None: privileged = False # Retrieve the sysacctmanagersdns and see if the DN is there try: entry = ldap.get_entry( sysaccount_mgrs_dn, [attr_sysacctmgrdns] ) managers = entry.get(attr_sysacctmgrdns, []) if str(dn) in managers: privileged = True except errors.NotFound: pass entry_attrs['privileged'] = privileged if exc is not None: raise exc @register() class sysaccount_add(LDAPCreate): __doc__ = _('Add a new IPA system account.') msg_summary = _('Added system account "%(value)s"') takes_options = LDAPCreate.takes_options + update_without_reset has_output_params = LDAPCreate.has_output_params + update_without_reset def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys, **options): assert isinstance(dn, DN) if 'userpassword' not in entry_attrs and 'random' not in options: raise errors.ValidationError( name='password', error=_('Either --password or --random is required') ) check_userpassword(entry_attrs, **options) validate_nsaccountlock(entry_attrs) return dn def post_callback(self, ldap, dn, entry_attrs, *keys, **options): assert isinstance(dn, DN) fill_randompassword(entry_attrs, **options) try: self.obj.handle_reset(self, self.api.Command.sysaccount_policy, ldap, dn, entry_attrs, **options) except errors.NotGroupMember: pass self.add_message(SystemAccountUsage(uid=keys[0], dn=dn)) convert_nsaccountlock(entry_attrs) return dn @register() class sysaccount_del(LDAPDelete): __doc__ = _('Delete an IPA system account.') msg_summary = _('Deleted system account "%(value)s"') def pre_callback(self, ldap, dn, *keys, **options): assert isinstance(dn, DN) sysaccount = keys[-1] if sysaccount.lower() in required_system_accounts: raise errors.ValidationError( name='system account', error=_('{} is required by the IPA master').format(sysaccount) ) # Make sure to remove the sysaccount entry from passsync_mgrs_dn # don't error out if access is denied try: options['privileged'] = False self.obj.handle_reset(self, None, ldap, dn, None, **options) except (errors.ACIError, errors.NotGroupMember): pass return dn @register() class sysaccount_mod(LDAPUpdate): __doc__ = _('Modify an existing IPA system account.') takes_options = LDAPUpdate.takes_options + update_without_reset has_output_params = LDAPUpdate.has_output_params + update_without_reset allow_empty_update = True msg_summary = _('Modified service "%(value)s"') def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys, **options): assert isinstance(dn, DN) check_userpassword(entry_attrs, **options) try: self.obj.handle_reset(self, self.api.Command.sysaccount_policy, ldap, dn, entry_attrs, **options) except (errors.EmptyModlist, errors.NotGroupMember, errors.AlreadyGroupMember): object.__setattr__(self, 'allow_empty_update', False) setattr(context, 'privileged', entry_attrs['privileged']) del entry_attrs['privileged'] if 'privileged' not in options: object.__setattr__(self, 'allow_empty_update', False) validate_nsaccountlock(entry_attrs) return dn def post_callback(self, ldap, dn, entry_attrs, *keys, **options): assert isinstance(dn, DN) fill_randompassword(entry_attrs, **options) entry_attrs['privileged'] = getattr(context, 'privileged') convert_nsaccountlock(entry_attrs) return dn @register() class sysaccount_find(LDAPSearch): __doc__ = _('Search for IPA system accounts.') msg_summary = ngettext( '%(count)d system account matched', '%(count)d system accounts matched', 0 ) sort_result_entries = False takes_options = LDAPSearch.takes_options has_output_params = LDAPSearch.has_output_params + update_without_reset def post_callback(self, ldap, entries, truncated, *args, **options): if options.get('pkey_only', False): return truncated for entry_attrs in entries: self.obj.get_password_attributes(ldap, entry_attrs.dn, entry_attrs) self.obj.handle_reset(self, self, ldap, entry_attrs.dn, entry_attrs, **options) convert_nsaccountlock(entry_attrs) return truncated @register() class sysaccount_show(LDAPRetrieve): __doc__ = _('Display information about an IPA system account.') member_attributes = ['memberof'] has_output_params = LDAPRetrieve.has_output_params + update_without_reset def post_callback(self, ldap, dn, entry_attrs, *keys, **options): assert isinstance(dn, DN) self.obj.get_password_attributes(ldap, dn, entry_attrs) self.obj.handle_reset(self, self, ldap, dn, entry_attrs, **options) convert_nsaccountlock(entry_attrs) return dn @register() class sysaccount_policy(LDAPRetrieve): __doc__ = _( 'Manage the system account policy.' ) takes_options = LDAPRetrieve.takes_options + update_without_reset has_output_params = LDAPRetrieve.has_output_params + update_without_reset def post_callback(self, ldap, dn, entry_attrs, *keys, **options): self.obj.handle_reset(self, self, ldap, dn, entry_attrs, **options) convert_nsaccountlock(entry_attrs) return dn @register() class sysaccount_disable(LDAPQuery): __doc__ = _('Disable a system account.') has_output = output.standard_value msg_summary = _('Disabled system account "%(value)s"') def execute(self, *keys, **options): ldap = self.obj.backend dn = self.obj.get_dn(*keys, **options) ldap.deactivate_entry(dn) return dict( result=True, value=pkey_to_value(keys[0], options), ) @register() class sysaccount_enable(LDAPQuery): __doc__ = _('Enable a system account.') has_output = output.standard_value has_output_params = LDAPQuery.has_output_params msg_summary = _('Enabled system account "%(value)s"') def execute(self, *keys, **options): ldap = self.obj.backend dn = self.obj.get_dn(*keys, **options) ldap.activate_entry(dn) return dict( result=True, value=pkey_to_value(keys[0], options), )