mirror of
https://salsa.debian.org/freeipa-team/freeipa.git
synced 2024-12-29 10:21:18 -06:00
292 lines
9.0 KiB
Python
292 lines
9.0 KiB
Python
# Authors:
|
|
# Jason Gerard DeRose <jderose@redhat.com>
|
|
# Rob Crittenden <rcritten@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; version 2 only
|
|
#
|
|
# 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, write to the Free Software
|
|
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
|
|
|
|
"""
|
|
Frontend plugins for service (Identity).
|
|
"""
|
|
|
|
from ipalib import api, crud, errors
|
|
from ipalib import Object # Plugin base classes
|
|
from ipalib import Str, Flag, Bytes # Parameter types
|
|
import base64
|
|
from OpenSSL import crypto
|
|
|
|
default_attributes = ['krbprincipalname', 'usercertificate']
|
|
|
|
def validate_principal(ugettext, principal):
|
|
(service, hostname, principal) = split_principal(principal)
|
|
|
|
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]
|
|
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 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 = service + "/" + hostname + "@" + realm
|
|
|
|
return unicode(principal)
|
|
|
|
def validate_certificate(ugettext, cert):
|
|
"""
|
|
For now just verify that it is properly base64-encoded.
|
|
"""
|
|
try:
|
|
base64.b64decode(cert)
|
|
except Exception, e:
|
|
raise errors.Base64DecodeError(reason=str(e))
|
|
|
|
class service(Object):
|
|
"""
|
|
Service object.
|
|
"""
|
|
takes_params = (
|
|
Str('principal',
|
|
validate_principal,
|
|
primary_key=True,
|
|
normalizer=lambda value: normalize_principal(value),
|
|
),
|
|
Bytes('usercertificate?',
|
|
validate_certificate,
|
|
cli_name='certificate',
|
|
doc='Base-64 encoded server certificate',
|
|
),
|
|
)
|
|
api.register(service)
|
|
|
|
|
|
class service_add(crud.Add):
|
|
"""
|
|
Add a new service.
|
|
"""
|
|
|
|
takes_options = (
|
|
Flag('force',
|
|
doc='Force a service principal name even if not found 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.ldap
|
|
|
|
force = kw.get('force', False)
|
|
try:
|
|
del kw['force']
|
|
except:
|
|
pass
|
|
|
|
(service, hostname, realm) = split_principal(principal)
|
|
|
|
if service.lower() == "host" and not force:
|
|
raise errors.HostService()
|
|
|
|
"""
|
|
FIXME once DNS client is done
|
|
if not force:
|
|
fqdn = hostname + "."
|
|
rs = dnsclient.query(fqdn, dnsclient.DNS_C_IN, dnsclient.DNS_T_A)
|
|
if len(rs) == 0:
|
|
self.log.debug("IPA: DNS A record lookup failed for '%s'" % hostname)
|
|
raise ipaerror.gen_exception(ipaerror.INPUT_NOT_DNS_A_RECORD)
|
|
else:
|
|
self.log.debug("IPA: found %d records for '%s'" % (len(rs), hostname))
|
|
"""
|
|
|
|
# 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'])
|
|
|
|
dn = ldap.make_service_dn(principal)
|
|
|
|
kw['dn'] = dn
|
|
kw['objectclass'] = ['krbPrincipal', 'krbPrincipalAux', 'krbTicketPolicyAux', 'ipaService', 'pkiUser']
|
|
|
|
return ldap.create(**kw)
|
|
|
|
def output_for_cli(self, textui, result, *args, **options):
|
|
textui.print_plain("Service added")
|
|
textui.print_entry(result)
|
|
|
|
api.register(service_add)
|
|
|
|
|
|
class service_del(crud.Del):
|
|
'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.ldap
|
|
dn = ldap.find_entry_dn("krbprincipalname", principal, object_type="ipaService")
|
|
entry = ldap.retrieve(dn)
|
|
if entry.has_key('usercertificate'):
|
|
cert = entry.get('usercertificate')
|
|
x509 = crypto.load_certificate(crypto.FILETYPE_ASN1, cert)
|
|
serial = str(x509.get_serial_number())
|
|
api.Command['cert_revoke'](unicode(serial, ), **{'revocation_reason': 5})
|
|
return ldap.delete(dn)
|
|
|
|
def output_to_cli(self, ret):
|
|
textui.print_plain("Service removed")
|
|
|
|
api.register(service_del)
|
|
|
|
class service_mod(crud.Update):
|
|
'Update an existing service.'
|
|
|
|
def execute(self, principal, **kw):
|
|
ldap = self.api.Backend.ldap
|
|
dn = ldap.find_entry_dn("krbprincipalname", principal, object_type="ipaService")
|
|
|
|
entry = ldap.retrieve(dn)
|
|
if entry.get('usercertificate') and kw.get('usercertificate'):
|
|
# FIXME, what to do here? Do we revoke the old cert?
|
|
raise errors.GenericError(format='entry already has a certificate')
|
|
|
|
# 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'])
|
|
|
|
return ldap.update(dn, **kw)
|
|
|
|
def output_to_cli(self, ret):
|
|
textui.print_plain("Service updated")
|
|
textui.print_entry(result)
|
|
|
|
api.register(service_mod)
|
|
|
|
|
|
class service_find(crud.Search):
|
|
'Search the existing services.'
|
|
takes_options = (
|
|
Flag('all', doc='Retrieve all attributes'),
|
|
)
|
|
def execute(self, principal, **kw):
|
|
ldap = self.api.Backend.ldap
|
|
|
|
search_kw = {}
|
|
search_kw['filter'] = "&(objectclass=ipaService)(!(objectClass=posixAccount))(!(|(krbprincipalname=kadmin/*)(krbprincipalname=K/M@*)(krbprincipalname=krbtgt/*)))"
|
|
search_kw['krbprincipalname'] = principal
|
|
|
|
object_type = ldap.get_object_type("krbprincipalname")
|
|
if object_type and not kw.get('objectclass'):
|
|
search_kw['objectclass'] = object_type
|
|
|
|
if kw.get('all', False):
|
|
search_kw['attributes'] = ['*']
|
|
else:
|
|
search_kw['attributes'] = default_attributes
|
|
|
|
return ldap.search(**search_kw)
|
|
|
|
def output_for_cli(self, textui, result, *args, **options):
|
|
counter = result[0]
|
|
services = result[1:]
|
|
if counter == 0:
|
|
textui.print_plain("No entries found")
|
|
return
|
|
|
|
for s in services:
|
|
textui.print_entry(s)
|
|
|
|
if counter == -1:
|
|
textui.print_plain("These results are truncated.")
|
|
textui.print_plain("Please refine your search and try again.")
|
|
textui.print_count(services, '%d services matched')
|
|
|
|
api.register(service_find)
|
|
|
|
|
|
class service_show(crud.Get):
|
|
'Examine an existing service.'
|
|
takes_options = (
|
|
Flag('all', doc='Display all service 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.ldap
|
|
dn = ldap.find_entry_dn("krbprincipalname", principal, object_type="ipaService")
|
|
# FIXME: should kw contain the list of attributes to display?
|
|
if kw.get('all', False):
|
|
return ldap.retrieve(dn)
|
|
else:
|
|
return ldap.retrieve(dn, default_attributes)
|
|
|
|
def output_for_cli(self, textui, result, *args, **options):
|
|
textui.print_entry(result)
|
|
|
|
api.register(service_show)
|