IntenseWebs/freeipa
3
0
Fork 0
mirror of https://salsa.debian.org/freeipa-team/freeipa.git synced 2024-12-28 18:01:23 -06:00
Code Issues Packages Projects Releases Wiki Activity
c799f6a0bf
freeipa/ipalib/plugins/service.py
Rob Crittenden 8d00d7c130 Enforce sizelimit in permission-find, post_callback returns truncated 
We actually perform two searches in permission-find. The first looks
for matches within the permission object itself. The second looks at
matches in the underlying aci.

We need to break out in two places. The first is if we find enough
matches in the permission itself. The second when we are appending
matches from acis.

The post_callback() definition needed to be modified to return
the truncated value so a plugin author can modify that value.

https://fedorahosted.org/freeipa/ticket/2322
2012-05-30 08:46:21 +02:00

523 lines
18 KiB
Python
Raw Blame History

# Authors:
# Jason Gerard DeRose <jderose@redhat.com>
# Rob Crittenden <rcritten@redhat.com>
# Pavel Zuna <pzuna@redhat.com>
#
# Copyright (C) 2008 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, either version 3 of the License, or
# (at your option) any later version.
#
# 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, see <http://www.gnu.org/licenses/>.
import base64
import os
from ipalib import api, errors, util
from ipalib import Str, Flag, Bytes
from ipalib.plugins.baseldap import *
from ipalib import x509
from ipalib import _, ngettext
from ipalib import util
import nss.nss as nss
from nss.error import NSPRError
from ipapython.ipautil import file_exists
__doc__ = _("""
Services
A IPA service represents a service that runs on a host. The IPA service
record can store a Kerberos principal, an SSL certificate, or both.
An IPA service can be managed directly from a machine, provided that
machine has been given the correct permission. This is true even for
machines other than the one the service is associated with. For example,
requesting an SSL certificate using the host service principal credentials
of the host. To manage a service using host credentials you need to
kinit as the host:
# kinit -kt /etc/krb5.keytab host/ipa.example.com@EXAMPLE.COM
Adding an IPA service allows the associated service to request an SSL
certificate or keytab, but this is performed as a separate step; they
are not produced as a result of adding the service.
Only the public aspect of a certificate is stored in a service record;
the private key is not stored.
EXAMPLES:
Add a new IPA service:
ipa service-add HTTP/web.example.com
Allow a host to manage an IPA service certificate:
ipa service-add-host --hosts=web.example.com HTTP/web.example.com
ipa role-add-member --hosts=web.example.com certadmin
Delete an IPA service:
ipa service-del HTTP/web.example.com
Find all IPA services associated with a host:
ipa service-find web.example.com
Find all HTTP services:
ipa service-find HTTP
Disable the service Kerberos key and SSL certificate:
ipa service-disable HTTP/web.example.com
Request a certificate for an IPA service:
ipa cert-request --principal=HTTP/web.example.com example.csr
Generate and retrieve a keytab for an IPA service:
ipa-getkeytab -s ipa.example.com -p HTTP/web.example.com -k /etc/httpd/httpd.keytab
""")
output_params = (
Flag('has_keytab',
label=_('Keytab'),
),
Str('managedby_host',
label='Managed by',
),
Str('subject',
label=_('Subject'),
),
Str('serial_number',
label=_('Serial Number'),
),
Str('serial_number_hex',
label=_('Serial Number (hex)'),
),
Str('issuer',
label=_('Issuer'),
),
Str('valid_not_before',
label=_('Not Before'),
),
Str('valid_not_after',
label=_('Not After'),
),
Str('md5_fingerprint',
label=_('Fingerprint (MD5)'),
),
Str('sha1_fingerprint',
label=_('Fingerprint (SHA1)'),
),
Str('revocation_reason?',
label=_('Revocation reason'),
)
)
def split_principal(principal):
service = hostname = realm = None
# Break down the principal into its component parts, which may or
# may not include the realm.
sp = principal.split('/')
if len(sp) != 2:
raise errors.MalformedServicePrincipal(reason='missing service')
service = sp[0]
if len(service) == 0:
raise errors.MalformedServicePrincipal(
reason='blank service'
)
sr = sp[1].split('@')
if len(sr) > 2:
raise errors.MalformedServicePrincipal(
reason='unable to determine realm'
)
hostname = sr[0].lower()
if len(sr) == 2:
realm = sr[1].upper()
# At some point we'll support multiple realms
if realm != api.env.realm:
raise errors.RealmMismatch()
else:
realm = api.env.realm
# Note that realm may be None.
return (service, hostname, realm)
def validate_principal(ugettext, principal):
(service, hostname, principal) = split_principal(principal)
return None
def normalize_principal(principal):
# The principal is already validated when it gets here
(service, hostname, realm) = split_principal(principal)
# Put the principal back together again
principal = '%s/%s@%s' % (service, hostname, realm)
return unicode(principal)
def validate_certificate(ugettext, cert):
"""
For now just verify that it is properly base64-encoded.
"""
if cert and util.isvalid_base64(cert):
try:
base64.b64decode(cert)
except Exception, e:
raise errors.Base64DecodeError(reason=str(e))
else:
# We'll assume this is DER data
pass
def set_certificate_attrs(entry_attrs):
"""
Set individual attributes from some values from a certificate.
entry_attrs is a dict of an entry
returns nothing
"""
if not 'usercertificate' in entry_attrs:
return
if type(entry_attrs['usercertificate']) in (list, tuple):
cert = entry_attrs['usercertificate'][0]
else:
cert = entry_attrs['usercertificate']
cert = x509.normalize_certificate(cert)
cert = x509.load_certificate(cert, datatype=x509.DER)
entry_attrs['subject'] = unicode(cert.subject)
entry_attrs['serial_number'] = unicode(cert.serial_number)
entry_attrs['serial_number_hex'] = u'0x%X' % cert.serial_number
entry_attrs['issuer'] = unicode(cert.issuer)
entry_attrs['valid_not_before'] = unicode(cert.valid_not_before_str)
entry_attrs['valid_not_after'] = unicode(cert.valid_not_after_str)
entry_attrs['md5_fingerprint'] = unicode(nss.data_to_hex(nss.md5_digest(cert.der_data), 64)[0])
entry_attrs['sha1_fingerprint'] = unicode(nss.data_to_hex(nss.sha1_digest(cert.der_data), 64)[0])
def check_required_principal(ldap, hostname, service):
"""
Raise an error if the host of this prinicipal is an IPA master and one
of the principals required for proper execution.
"""
try:
host_is_master(ldap, hostname)
except errors.ValidationError, e:
service_types = ['HTTP', 'ldap', 'DNS', 'dogtagldap']
if service in service_types:
raise errors.ValidationError(name='principal', error=_('This principal is required by the IPA master'))
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'
]
search_attributes = ['krbprincipalname', 'managedby']
default_attributes = ['krbprincipalname', 'usercertificate', 'managedby']
uuid_attribute = 'ipauniqueid'
attribute_members = {
'managedby': ['host'],
}
bindable = True
relationships = {
'managedby': ('Managed by', 'man_by_', 'not_man_by_'),
}
password_attributes = [('krbprincipalkey', 'has_keytab')]
label = _('Services')
label_singular = _('Service')
takes_params = (
Str('krbprincipalname', validate_principal,
cli_name='principal',
label=_('Principal'),
doc=_('Service principal'),
primary_key=True,
normalizer=lambda value: normalize_principal(value),
),
Bytes('usercertificate?', validate_certificate,
cli_name='certificate',
label=_('Certificate'),
doc=_('Base-64 encoded server certificate'),
flags=['no_search',],
)
)
api.register(service)
class service_add(LDAPCreate):
__doc__ = _('Add a new IPA new service.')
msg_summary = _('Added service "%(value)s"')
member_attributes = ['managedby']
has_output_params = LDAPCreate.has_output_params + output_params
takes_options = (
Flag('force',
label=_('Force'),
doc=_('force principal name even if not in DNS'),
),
)
def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys, **options):
(service, hostname, realm) = split_principal(keys[-1])
if service.lower() == 'host' and not options['force']:
raise errors.HostService()
try:
hostresult = api.Command['host_show'](hostname)['result']
except errors.NotFound:
raise errors.NotFound(reason="The host '%s' does not exist to add a service to." % hostname)
cert = options.get('usercertificate')
if cert:
dercert = x509.normalize_certificate(cert)
x509.verify_cert_subject(ldap, hostname, dercert)
entry_attrs['usercertificate'] = dercert
if not options.get('force', False):
# We know the host exists if we've gotten this far but we
# really want to discourage creating services for hosts that
# don't exist in DNS.
util.validate_host_dns(self.log, hostname)
if not 'managedby' in entry_attrs:
entry_attrs['managedby'] = hostresult['dn']
return dn
api.register(service_add)
class service_del(LDAPDelete):
__doc__ = _('Delete an IPA service.')
msg_summary = _('Deleted service "%(value)s"')
member_attributes = ['managedby']
def pre_callback(self, ldap, dn, *keys, **options):
# In the case of services we don't want IPA master services to be
# deleted. This is a limited few though. If the user has their own
# custom services allow them to manage them.
(service, hostname, realm) = split_principal(keys[-1])
check_required_principal(ldap, hostname, service)
if self.api.env.enable_ra:
(dn, entry_attrs) = ldap.get_entry(dn, ['usercertificate'])
cert = entry_attrs.get('usercertificate')
if cert:
cert = cert[0]
try:
serial = unicode(x509.get_serial_number(cert, x509.DER))
try:
result = api.Command['cert_show'](unicode(serial))['result']
if 'revocation_reason' not in result:
try:
api.Command['cert_revoke'](unicode(serial), revocation_reason=4)
except errors.NotImplementedError:
# some CA's might not implement revoke
pass
except errors.NotImplementedError:
# some CA's might not implement revoke
pass
except NSPRError, nsprerr:
if nsprerr.errno == -8183:
# If we can't decode the cert them proceed with
# removing the service.
self.log.info("Problem decoding certificate %s" % nsprerr.args[1])
else:
raise nsprerr
return dn
api.register(service_del)
class service_mod(LDAPUpdate):
__doc__ = _('Modify an existing IPA service.')
msg_summary = _('Modified service "%(value)s"')
takes_options = LDAPUpdate.takes_options
has_output_params = LDAPUpdate.has_output_params + output_params
member_attributes = ['managedby']
def pre_callback(self, ldap, dn, entry_attrs, *keys, **options):
if 'usercertificate' in options:
(service, hostname, realm) = split_principal(keys[-1])
cert = options.get('usercertificate')
if cert:
dercert = x509.normalize_certificate(cert)
x509.verify_cert_subject(ldap, hostname, dercert)
(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' % (
x509.get_serial_number(entry_attrs_old['usercertificate'][0], x509.DER)
)
raise errors.GenericError(format=fmt)
entry_attrs['usercertificate'] = dercert
else:
entry_attrs['usercertificate'] = None
return dn
def post_callback(self, ldap, dn, entry_attrs, *keys, **options):
set_certificate_attrs(entry_attrs)
api.register(service_mod)
class service_find(LDAPSearch):
__doc__ = _('Search for IPA services.')
msg_summary = ngettext(
'%(count)d service matched', '%(count)d services matched', 0
)
member_attributes = ['managedby']
takes_options = LDAPSearch.takes_options
has_output_params = LDAPSearch.has_output_params + output_params
def pre_callback(self, ldap, filter, attrs_list, base_dn, scope, *args, **options):
# lisp style!
custom_filter = '(&(objectclass=ipaService)' \
'(!(objectClass=posixAccount))' \
'(!(|(krbprincipalname=kadmin/*)' \
'(krbprincipalname=K/M@*)' \
'(krbprincipalname=krbtgt/*))' \
')' \
')'
return (
ldap.combine_filters((custom_filter, filter), rules=ldap.MATCH_ALL),
base_dn, scope
)
def post_callback(self, ldap, entries, truncated, *args, **options):
if options.get('pkey_only', False):
return truncated
for entry in entries:
(dn, entry_attrs) = entry
self.obj.get_password_attributes(ldap, dn, entry_attrs)
set_certificate_attrs(entry_attrs)
return truncated
api.register(service_find)
class service_show(LDAPRetrieve):
__doc__ = _('Display information about an IPA service.')
member_attributes = ['managedby']
takes_options = LDAPRetrieve.takes_options + (
Str('out?',
doc=_('file to store certificate in'),
),
)
has_output_params = LDAPRetrieve.has_output_params + output_params
def post_callback(self, ldap, dn, entry_attrs, *keys, **options):
self.obj.get_password_attributes(ldap, dn, entry_attrs)
set_certificate_attrs(entry_attrs)
return dn
def forward(self, *keys, **options):
if 'out' in options:
util.check_writable_file(options['out'])
result = super(service_show, self).forward(*keys, **options)
if 'usercertificate' in result['result']:
x509.write_certificate(result['result']['usercertificate'][0], options['out'])
result['summary'] = _('Certificate stored in file \'%(file)s\'') % dict(file=options['out'])
return result
else:
raise errors.NoCertificateError(entry=keys[-1])
else:
return super(service_show, self).forward(*keys, **options)
api.register(service_show)
class service_add_host(LDAPAddMember):
__doc__ = _('Add hosts that can manage this service.')
member_attributes = ['managedby']
has_output_params = LDAPAddMember.has_output_params + output_params
api.register(service_add_host)
class service_remove_host(LDAPRemoveMember):
__doc__ = _('Remove hosts that can manage this service.')
member_attributes = ['managedby']
has_output_params = LDAPRemoveMember.has_output_params + output_params
api.register(service_remove_host)
class service_disable(LDAPQuery):
__doc__ = _('Disable the Kerberos key and SSL certificate of a service.')
has_output = output.standard_value
msg_summary = _('Disabled service "%(value)s"')
has_output_params = LDAPQuery.has_output_params + output_params
def execute(self, *keys, **options):
ldap = self.obj.backend
dn = self.obj.get_dn(*keys, **options)
(dn, entry_attrs) = ldap.get_entry(dn, ['usercertificate'])
(service, hostname, realm) = split_principal(keys[-1])
check_required_principal(ldap, hostname, service)
# See if we do any work at all here and if not raise an exception
done_work = False
if 'usercertificate' in entry_attrs:
cert = x509.normalize_certificate(entry_attrs.get('usercertificate')[0])
try:
serial = unicode(x509.get_serial_number(cert, x509.DER))
try:
result = api.Command['cert_show'](unicode(serial))['result']
if 'revocation_reason' not in result:
try:
api.Command['cert_revoke'](unicode(serial), revocation_reason=4)
except errors.NotImplementedError:
# some CA's might not implement revoke
pass
except errors.NotImplementedError:
# some CA's might not implement revoke
pass
except NSPRError, nsprerr:
if nsprerr.errno == -8183:
# If we can't decode the cert them proceed with
# disabling the service
self.log.info("Problem decoding certificate %s" % nsprerr.args[1])
else:
raise nsprerr
# Remove the usercertificate altogether
ldap.update_entry(dn, {'usercertificate': None})
done_work = True
self.obj.get_password_attributes(ldap, dn, entry_attrs)
if entry_attrs['has_keytab']:
ldap.remove_principal_key(dn)
done_work = True
if not done_work:
raise errors.AlreadyInactive()
return dict(
result=True,
value=keys[0],
)
api.register(service_disable)
Reference in New Issue View Git Blame Copy Permalink