diff options
| -rw-r--r-- | ipalib/plugins/service.py | 271 |
1 files changed, 66 insertions, 205 deletions
diff --git a/ipalib/plugins/service.py b/ipalib/plugins/service.py index 70812579d..782948a5a 100644 --- a/ipalib/plugins/service.py +++ b/ipalib/plugins/service.py @@ -25,24 +25,22 @@ import base64 from OpenSSL import crypto -from ipalib import api, crud, errors -from ipalib import Object +from ipalib import api, errors from ipalib import Str, Flag, Bytes -from ipalib import uuid +from ipalib.plugins.baseldap import * -_container_dn = api.env.container_service -_default_attributes = ['krbprincipalname', 'usercertificate'] def get_serial(certificate): """ - Given a certificate, return the serial number in that cert + Given a certificate, return the serial number in that cert. """ try: x509 = crypto.load_certificate(crypto.FILETYPE_ASN1, certificate) serial = str(x509.get_serial_number()) except crypto.Error: - raise errors.GenericError(format='Unable to decode certificate in entry') - + raise errors.GenericError( + format='Unable to decode certificate in entry' + ) return serial def split_principal(principal): @@ -94,60 +92,63 @@ def validate_certificate(ugettext, cert): raise errors.Base64DecodeError(reason=str(e)) -class service(Object): +class service(LDAPObject): """ Service object. """ + container_dn = api.env.container_service + object_name = 'service' + object_name_plural = 'services' + object_class = [ + 'krbprincipal', 'krbprincipalaux', 'krbticketpolicyaux', 'ipaobject', + 'ipaservice', 'pkiuser' + ] + default_attributes = ['krbprincipalname', 'usercertificate'] + uuid_attribute = 'ipauniqueid' + attribute_names = { + 'krbprincipalname': 'kerberos principal', + 'usercertificate': 'user certificate', + 'ipauniqueid': 'unique identifier', + } + takes_params = ( Str('krbprincipalname', validate_principal, cli_name='principal', - doc='Service principal', + doc='service principal', primary_key=True, normalizer=lambda value: normalize_principal(value), ), Bytes('usercertificate?', validate_certificate, cli_name='certificate', - doc='Base-64 encoded server certificate', + doc='base-64 encoded server certificate', ), ) api.register(service) -class service_add(crud.Create): +class service_add(LDAPCreate): """ Add new service. """ takes_options = ( Flag('force', - doc='Force principal name even if not in DNS', + doc='force principal name even if not in DNS', ), ) - def execute(self, principal, **kw): - """ - Execute the service-add operation. - - The dn should not be passed as a keyword argument as it is constructed - by this method. - - Returns the entry as it will be created in LDAP. - - :param principal: The service to be added in the form: service/hostname - :param kw: Keyword arguments for the other LDAP attributes. - """ - assert 'krbprincipalname' not in kw - ldap = self.api.Backend.ldap2 - # FIXME: should be in a normalizer. Need to fix normalizers to work - # on non-unicode data - if kw.get('usercertificate'): - kw['usercertificate'] = base64.b64decode(kw['usercertificate']) - - (service, hostname, realm) = split_principal(principal) - - if service.lower() == 'host' and not kw['force']: + def pre_callback(self, ldap, dn, entry_attrs, *keys, **options): + (service, hostname, realm) = split_principal(keys[-1]) + if service.lower() == 'host' and not options['force']: raise errors.HostService() - # FIXME: once DNS client is done + cert = entry_attrs.get('usercertificate') + if cert: + # FIXME: should be in a normalizer: need to fix normalizers + # to work on non-unicode data + entry_attrs['usercertificate'] = base64.b64decode(cert) + # FIXME: shouldn't we request signing at this point? + + # TODO: once DNS client is done (code below for reference only!) # if not kw['force']: # fqdn = hostname + '.' # rs = dnsclient.query(fqdn, dnsclient.DNS_C_IN, dnsclient.DNS_T_A) @@ -161,115 +162,53 @@ class service_add(crud.Create): # 'IPA: found %d records for '%s'" % (len(rs), hostname) # ) - entry_attrs = self.args_options_2_entry(principal, **kw) - entry_attrs['objectclass'] = [ - 'krbprincipal', 'krbprincipalaux', 'krbticketpolicyaux', - 'ipaobject', 'ipaservice', 'pkiuser' - ] - entry_attrs['ipauniqueid'] = str(uuid.uuid1()) - dn = ldap.make_dn(entry_attrs, 'krbprincipalname', _container_dn) - - ldap.add_entry(dn, entry_attrs) - - return ldap.get_entry(dn, entry_attrs.keys()) - - def output_for_cli(self, textui, result, principal, **options): - (dn, entry_attrs) = result - - textui.print_name(self.name) - textui.print_attribute('dn', dn) - textui.print_entry(entry_attrs) - textui.print_dashed('Created service "%s".' % principal) + return dn api.register(service_add) -class service_del(crud.Delete): +class service_del(LDAPDelete): """ Delete an existing service. """ - def execute(self, principal, **kw): - """ - Delete a service principal. - - principal is the krbprincipalname of the entry to delete. - - This should be called with much care. - - :param principal: The service to be added in the form: service/hostname - :param kw: not used - """ - ldap = self.api.Backend.ldap2 - - (dn, entry_attrs) = ldap.find_entry_by_attr( - 'krbprincipalname', principal, 'ipaservice' - ) - - if 'usercertificate' in entry_attrs: - serial = get_serial(entry_attrs['usercertificate']) - api.Command['cert_revoke'](unicode(serial), revocation_reason=5) - - ldap.delete_entry(dn) - - return True - - def output_for_cli(self, textui, result, principal, **options): - textui.print_name(self.name) - textui.print_dashed('Deleted service "%s".' % principal) + def pre_callback(self, ldap, dn, *keys, **options): + if self.api.env.enable_ra: + (dn, entry_attrs) = ldap.get_entry(dn, ['usercertificate']) + cert = entry_attrs.get('usercertificate') + if cert: + serial = unicode(get_serial(cert)) + self.api.Command['cert_revoke'](serial, revocation_reason=5) + return dn api.register(service_del) -class service_mod(crud.Update): +class service_mod(LDAPUpdate): """ Modify service. """ - def execute(self, principal, **options): - ldap = self.api.Backend.ldap2 - # FIXME, should be in a normalizer. Need to fix normalizers to work - # on non-unicode data. - if options.get('usercertificate'): - options['usercertificate'] = base64.b64decode(options['usercertificate']) - - entry_attrs = self.args_options_2_entry(*tuple(), **options) - dn = ldap.make_dn_from_attr('krbprincipalname', principal, _container_dn) - - (dn, old_entry_attrs) = ldap.get_entry(dn) - if 'usercertificate' in old_entry_attrs and 'usercertificate' in options: - # FIXME, what to do here? Do we revoke the old cert? - raise errors.GenericError(format='entry already has a certificate, serial number %s' % get_serial(old_entry_attrs['usercertificate'])) - - try: - ldap.update_entry(dn, entry_attrs) - except errors.EmptyModlist: - pass - - return ldap.get_entry(dn, entry_attrs.keys()) - - def output_to_cli(self, textui, result, principal, **options): - (dn, entry_attrs) = result - - textui.print_name(self.name) - textui.print_attribute('dn', dn) - textui.print_entry(entry_attrs) - textui.print_dashed('Modified service "%s".' % principal) + def pre_callback(self, ldap, dn, entry_attrs, *keys, **options): + cert = entry_attrs.get('usercertificate') + if cert: + (dn, entry_attrs_old) = ldap.get_entry(dn, ['usercertificate']) + if 'usercertificate' in entry_attrs_old: + # FIXME: what to do here? do we revoke the old cert? + fmt = 'entry already has a certificate, serial number: %s' % ( + get_serial(entry_attrs_old['usercertificate']) + ) + raise errors.GenericError(format=fmt) + # FIXME: should be in normalizer; see service_add + entry_attrs['usercertificate'] = base64.b64decode(cert) + return dn api.register(service_mod) -class service_find(crud.Search): +class service_find(LDAPSearch): """ Search for services. """ - takes_options = ( - Flag('all', - doc='Retrieve all attributes' - ), - ) - - def execute(self, term, **kw): - ldap = self.api.Backend.ldap2 - + def pre_callback(self, ldap, filter, attrs_list, base_dn, *args, **options): # lisp style! custom_filter = '(&(objectclass=ipaService)' \ '(!(objectClass=posixAccount))' \ @@ -278,95 +217,17 @@ class service_find(crud.Search): '(krbprincipalname=krbtgt/*))' \ ')' \ ')' - - search_kw = self.args_options_2_entry(**kw) - search_kw['objectclass'] = 'krbprincipal' - filter = ldap.make_filter(search_kw, rules=ldap.MATCH_ALL) - - search_kw = {} - for a in _default_attributes: - search_kw[a] = term - term_filter = ldap.make_filter(search_kw, exact=False) - - filter = ldap.combine_filters( - (custom_filter, filter, term_filter), rules=ldap.MATCH_ALL + return ldap.combine_filters( + (custom_filter, filter), rules=ldap.MATCH_ALL ) - if kw['all']: - attrs_list = ['*'] - else: - attrs_list = _default_attributes - - try: - (entries, truncated) = ldap.find_entries( - filter, attrs_list, _container_dn - ) - except errors.NotFound: - (entries, truncated) = (tuple(), False) - - return (entries, truncated) - - def output_for_cli(self, textui, result, principal, **options): - (entries, truncated) = result - - textui.print_name(self.name) - for (dn, entry_attrs) in entries: - textui.print_attribute('dn', dn) - textui.print_entry(entry_attrs) - textui.print_plain('') - textui.print_count( - len(entries), '%i service matched.', '%i services matched.' - ) - if truncated: - textui.print_dashed('These results are truncated.', below=False) - textui.print_dashed( - 'Please refine your search and try again.', above=False - ) - api.register(service_find) -class service_show(crud.Retrieve): +class service_show(LDAPRetrieve): """ Display service. """ - takes_options = ( - Flag('all', - doc='Retrieve all attributes' - ), - ) - - def execute(self, principal, **kw): - """ - Execute the service-show operation. - - The dn should not be passed as a keyword argument as it is constructed - by this method. - - Returns the entry - - :param principal: The service principal to retrieve - :param kw: Not used. - """ - ldap = self.api.Backend.ldap2 - - dn = ldap.make_dn_from_attr( - 'krbprincipalname', principal, _container_dn - ) - - if kw['all']: - attrs_list = ['*'] - else: - attrs_list = _default_attributes - - return ldap.get_entry(dn, attrs_list) - - def output_for_cli(self, textui, result, principal, **options): - (dn, entry_attrs) = result - - textui.print_name(self.name) - textui.print_attribute('dn', dn) - textui.print_entry(entry_attrs) api.register(service_show) |