IntenseWebs/freeipa
3
0
Fork 0
mirror of https://salsa.debian.org/freeipa-team/freeipa.git synced 2025-02-25 18:55:28 -06:00
Code Issues Packages Projects Releases Wiki Activity

rebase dogtag clean-up patch

Browse Source
This commit is contained in:
John Dennis
2009-12-08 16:57:07 -05:00
committed by Jason Gerard DeRose
parent 62d40286ac
commit ee909d871c
6 changed files with 1741 additions and 291 deletions
Download Patch File Download Diff File Expand all files Collapse all files

131
checks/check-ra.py
View File

@@ -14,87 +14,98 @@ server. I don't exactly remember the steps, so ping him for help.
from os import path
import sys
parent = path.dirname(path.dirname(path.abspath(__file__)))
sys.path.append(parent)
sys.path.insert(0, parent)
verbose = True
from base64 import b64encode, b64decode
from ipalib import api
# certificate with serial number 17
cert = b64decode("""
MIIC3zCCAcegAwIBAgIBETANBgkqhkiG9w0BAQUFADA7MRkwFwYDVQQKExBTamNSZWRoYXQgRG9tYW
luMR4wHAYDVQQDExVDZXJ0aWZpY2F0ZSBBdXRob3JpdHkwHhcNMDkwMTIyMjMzODA2WhcNMDkwNzIx
MjMzODA2WjAUMRIwEAYKCZImiZPyLGQBARMCbGwwgZ8wDQYJKoZIhvcNAQEBBQADgY0AMIGJAoGBAM
id6i9ri9ldyAXaH4MJSPdUDjdc9+E10hwxw7crFE1K0uvr8YT2e1YotNqv7Q+Bk7KVRrLH6Y5UPlWY
uSAP8G9t8yjn5Uo3iXU5AqsrRek+pxerD/WocwedF6yjJ/zlQyYyg93h0njJr1lStyVLTyp+VVqtk3
FSDIwLCWQHOTejAgMBAAGjgZgwgZUwHwYDVR0jBBgwFoAUlz9JZxqVabh4QQOEkxyWt80pIQkwQwYI
KwYBBQUHAQEENzA1MDMGCCsGAQUFBzABhidodHRwOi8vYS1mOC5zamMucmVkaGF0LmNvbTo5MTgwL2
NhL29jc3AwDgYDVR0PAQH/BAQDAgXgMB0GA1UdJQQWMBQGCCsGAQUFBwMCBggrBgEFBQcDBDANBgkq
hkiG9w0BAQUFAAOCAQEAhU+oqPh+rlYFPm0D8HAJ0RIWw9gkNctHUfVGi+NeYTaUAEGWUOpXjLSQgP
gq1fNBHd+IRLhycwp4uUsFCPE1n3eStmn/D6o9u1eNnTFPj74MLZVQQTXPE8+LBYeHgTUwFuKp2WyW
9J/BDZ3pDWKYWWMawhD7ext7UhZkpIJODFEaDxiXCfB8GsAEbmfoYFk21znuGQQu3Wu1s6licyunLh
/W3sxCFGIT9DHxS0GZKimm7M02IPGxK/0TZr0kVcLQx6XGKqiK1464rvl4u60mQjwJwfhawshs84YT
xFnXZKkvsT3GjfIe/k687TMG3paTFtKkis+u7z0v6355uJzLpQ==
""")
subject = u'CN=vm-070.idm.lab.bos.redhat.com'
csr = '\
MIIBZzCB0QIBADAoMSYwJAYDVQQDEx12bS0wNzAuaWRtLmxhYi5ib3MucmVkaGF0\n\
LmNvbTCBnzANBgkqhkiG9w0BAQEFAAOBjQAwgYkCgYEAriTSlAG+/xkvtxliWMeO\n\
Qu+vFQTz+/fgy7xWIg6WR2At6j/9eJ7LUYhqguqevOAQpuePxY4/FEfpmQ6PTgs/\n\
LXKa0vhIkXzkmMjKynUIWHYeaZekcXxye1dV/PdNB6H801xs60YjbScOJj3Hexvm\n\
hOKsdmwO1ukqTTEKDXrr3c8CAwEAAaAAMA0GCSqGSIb3DQEBBQUAA4GBAG4pTLrE\n\
cvrkQXVdMOjgAVJ6KZYl/caIOYhIlcJ3jhf95Bv/Zs3lpfHjXnM8jj4EWfyd0lZx\n\
2EUytXXubKJUpjUCeBp4oaQ2Ahvdxo++oUcbXkKxtCOUB6Mw8XEIVYaldZlcHDHM\n\
dysLdrZ3K9HOzoeSq2e0m+trQaWnBQG47O7F\n\
'
csr = 'MIIBlDCB/gIBADBVMR0wGwYDVQQKExRVc2Vyc3lzUmVkaGF0LURvbWFpbjEQMA4GA1UECxMHcGtpLWlwYTEiMCAGA1UEAxMZSVBBLVN1YnN5c3RlbS1DZXJ0aWZpY2F0ZTCBnzANBgkqhkiG9w0BAQEFAAOBjQAwgYkCgYEA3Qmpr81WxbnISmyyhc2ShiPzUvWIrCg5FgJ1QrBl7CRe62Wl/YYiV/DbuMoex1ec7zKfgfSFVFU9/2iwj7Du0sZdXYJNQPdj9yLdPk2tyxdgJuHLdxI0SNgaEFyvmIMP/X9vQN9H5w0/PyrJQscOxc6tbTcYL0ZSSylLQ+diaQECAwEAA'
reference_decode = {
'certificate' : b64decode
}
trial_decode = {
'certificate' : b64decode
}
api.bootstrap(
in_server=True,
enable_ra=True,
ca_host='a-f8.sjc.redhat.com',
ra_plugin='dogtag',
ca_host='vm-070.idm.lab.bos.redhat.com',
debug=True,
in_tree=True,
)
api.finalize()
ra = api.Backend.ra
def assert_equal(*vals):
val0 = vals[0]
for val in vals[1:]:
assert val == val0, '%r != %r' % (val, val0)
def assert_equal(trial, reference):
keys = reference.keys()
keys.sort()
for key in keys:
reference_val = reference[key]
trial_val = trial[key]
if reference_decode.has_key(key):
reference_val = reference_decode[key](reference_val)
api.log.info('******** Testing ra.check_request_status() ********')
assert_equal(
ra.check_request_status('35'),
dict(
status='0',
serial_number='17',
request_status='complete',
request_id='35',
)
)
if trial_decode.has_key(key):
trial_val = trial_decode[key](trial_val)
assert reference_val == trial_val, \
'%s: not equal\n\nreference_val:\n%r\ntrial_val:\n%r' % \
(key, reference[key], trial[key])
api.log.info('******** Testing ra.get_certificate() ********')
assert_equal(
ra.get_certificate('17'),
dict(
status='0',
certificate=b64encode(cert),
)
)
api.log.info('******** Testing ra.request_certificate() ********')
assert_equal(
ra.request_certificate(csr),
dict(
status='1',
)
)
request_result = ra.request_certificate(csr)
if verbose: print "request_result=\n%s" % request_result
assert_equal(request_result,
{'subject' : subject,
})
api.log.info('******** Testing ra.check_request_status() ********')
status_result = ra.check_request_status(request_result['request_id'])
if verbose: print "status_result=\n%s" % status_result
assert_equal(status_result,
{'serial_number' : request_result['serial_number'],
'request_id' : request_result['request_id'],
'cert_request_status' : u'complete',
})
api.log.info('******** Testing ra.get_certificate() ********')
get_result = ra.get_certificate(request_result['serial_number'])
if verbose: print "get_result=\n%s" % get_result
assert_equal(get_result,
{'serial_number' : request_result['serial_number'],
'certificate' : request_result['certificate'],
})
api.log.info('******** Testing ra.revoke_certificate() ********')
assert_equal(
ra.revoke_certificate('17', revocation_reason=6), # Put on hold
dict(
status='0',
revoked=True,
)
)
revoke_result = ra.revoke_certificate(request_result['serial_number'],
revocation_reason=6) # Put on hold
if verbose: print "revoke_result=\n%s" % revoke_result
assert_equal(revoke_result,
{'revoked' : True
})
api.log.info('******** Testing ra.take_certificate_off_hold() ********')
assert_equal(
ra.take_certificate_off_hold('17'),
dict(
taken_off_hold=True,
)
)
unrevoke_result = ra.take_certificate_off_hold(request_result['serial_number'])
if verbose: print "unrevoke_result=\n%s" % unrevoke_result
assert_equal(unrevoke_result,
{'unrevoked' : True
})

36
ipalib/plugins/cert.py
View File

@@ -1,6 +1,7 @@
# Authors:
# Andrew Wnuk <awnuk@redhat.com>
# Jason Gerard DeRose <jderose@redhat.com>
# John Dennis <jdennis@redhat.com>
#
# Copyright (C) 2009 Red Hat
# see file 'COPYING' for use and warranty information
@@ -38,10 +39,12 @@ from ipapython import dnsclient
from pyasn1.error import PyAsn1Error
import logging
import traceback
from ipalib.request import ugettext as _
def get_serial(certificate):
"""
Given a certificate, return the serial number in that cert
as a Python long object.
In theory there should be only one cert per object so even if we get
passed in a list/tuple only return the first one.
@@ -49,9 +52,9 @@ def get_serial(certificate):
if type(certificate) in (list, tuple):
certificate = certificate[0]
try:
serial = str(x509.get_serial_number(certificate))
serial = x509.get_serial_number(certificate)
except PyAsn1Error:
raise errors.GenericError(format='Unable to decode certificate in entry')
raise errors.CertificateOperationError(error=_('Unable to decode certificate in entry'))
return serial
@@ -69,7 +72,7 @@ def get_csr_hostname(csr):
# The ASN.1 decoding errors tend to be long and involved and the
# last bit is generally not interesting. We need the whole traceback.
logging.error('Unable to decode CSR\n%s', traceback.format_exc())
raise errors.GenericError(format='Failure decoding Certificate Signing Request')
raise errors.CertificateOperationError(error=_('Failure decoding Certificate Signing Request'))
return None
@@ -83,7 +86,7 @@ def get_subjectaltname(csr):
# The ASN.1 decoding errors tend to be long and involved and the
# last bit is generally not interesting. We need the whole traceback.
logging.error('Unable to decode CSR\n%s', traceback.format_exc())
raise errors.GenericError(format='Failure decoding Certificate Signing Request')
raise errors.CertificateOperationError(error=_('Failure decoding Certificate Signing Request'))
return request.get_subjectaltname()
def validate_csr(ugettext, csr):
@@ -100,9 +103,9 @@ def validate_csr(ugettext, csr):
except TypeError, e:
raise errors.Base64DecodeError(reason=str(e))
except PyAsn1Error:
raise errors.GenericError(format='Failure decoding Certificate Signing Request')
raise errors.CertificateOperationError(error=_('Failure decoding Certificate Signing Request'))
except Exception, e:
raise errors.GenericError(format='Failure decoding Certificate Signing Request: %s' % str(e))
raise errors.CertificateOperationError(error=_('Failure decoding Certificate Signing Request: %s') % str(e))
class cert_request(VirtualCommand):
@@ -170,7 +173,7 @@ class cert_request(VirtualCommand):
(dn, service) = api.Command['service_show'](principal, all=True, raw=True)
if 'usercertificate' in service:
# 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(base64.b64encode(service['usercertificate'][0])))
raise errors.CertificateOperationError(error=_('entry already has a certificate, serial number %s') % get_serial(base64.b64encode(service['usercertificate'][0])))
except errors.NotFound, e:
if not add:
raise errors.NotFound(reason="The service principal for this request doesn't exist.")
@@ -213,7 +216,7 @@ class cert_request(VirtualCommand):
if isinstance(result, dict) and len(result) > 0:
textui.print_entry(result, 0)
else:
textui.print_plain('Failed to submit a certificate request.')
textui.print_plain(_('Failed to submit a certificate request.'))
api.register(cert_request)
@@ -235,7 +238,7 @@ class cert_status(VirtualCommand):
if isinstance(result, dict) and len(result) > 0:
textui.print_entry(result, 0)
else:
textui.print_plain('Failed to retrieve a request status.')
textui.print_plain(_('Failed to retrieve a request status.'))
api.register(cert_status)
@@ -245,7 +248,8 @@ class cert_get(VirtualCommand):
Retrieve an existing certificate.
"""
takes_args = ('serial_number')
takes_args = (Str('serial_number',
doc='serial number in decimal or if prefixed with 0x in hexadecimal'))
operation="retrieve certificate"
def execute(self, serial_number):
@@ -256,7 +260,7 @@ class cert_get(VirtualCommand):
if isinstance(result, dict) and len(result) > 0:
textui.print_entry(result, 0)
else:
textui.print_plain('Failed to obtain a certificate.')
textui.print_plain(_('Failed to obtain a certificate.'))
api.register(cert_get)
@@ -266,7 +270,8 @@ class cert_revoke(VirtualCommand):
Revoke a certificate.
"""
takes_args = ('serial_number')
takes_args = (Str('serial_number',
doc='serial number in decimal or if prefixed with 0x in hexadecimal'))
operation = "revoke certificate"
# FIXME: The default is 0. Is this really an Int param?
@@ -288,7 +293,7 @@ class cert_revoke(VirtualCommand):
if isinstance(result, dict) and len(result) > 0:
textui.print_entry(result, 0)
else:
textui.print_plain('Failed to revoke a certificate.')
textui.print_plain(_('Failed to revoke a certificate.'))
api.register(cert_revoke)
@@ -298,7 +303,8 @@ class cert_remove_hold(VirtualCommand):
Take a revoked certificate off hold.
"""
takes_args = ('serial_number')
takes_args = (Str('serial_number',
doc='serial number in decimal or if prefixed with 0x in hexadecimal'))
operation = "certificate remove hold"
def execute(self, serial_number, **kw):
@@ -309,6 +315,6 @@ class cert_remove_hold(VirtualCommand):
if isinstance(result, dict) and len(result) > 0:
textui.print_entry(result, 0)
else:
textui.print_plain('Failed to take a revoked certificate off hold.')
textui.print_plain(_('Failed to take a revoked certificate off hold.'))
api.register(cert_remove_hold)

5
ipalib/plugins/service.py
View File

@@ -32,12 +32,13 @@ from pyasn1.error import PyAsn1Error
def get_serial(certificate):
"""
Given a certificate, return the serial number in that cert.
Given a certificate, return the serial number in that
cert as a Python long object.
"""
if type(certificate) in (list, tuple):
certificate = certificate[0]
try:
serial = str(x509.get_serial_number(certificate, type=x509.DER))
serial = x509.get_serial_number(certificate, type=x509.DER)
except PyAsn1Error:
raise errors.GenericError(
format='Unable to decode certificate in entry'

7
ipalib/x509.py
View File

@@ -188,8 +188,9 @@ class Certificate(univ.Sequence):
return info.getComponentByName('subject')
def get_serial_number(self):
'return the serial number as a Python long object'
info = self.getComponentByName('tbsCertificate')
return info.getComponentByName('serialNumber')
return long(info.getComponentByName('serialNumber'))
# end of ASN.1 data structures
@@ -230,9 +231,7 @@ def get_subject_components(certificate, type=PEM):
def get_serial_number(certificate, type=PEM):
"""
Return the serial number of a certificate.
Returns an integer
Return the serial number of a certificate as a Python long object.
"""
x509cert = load_certificate(certificate, type)
return x509cert.get_serial_number()

1751
ipaserver/plugins/dogtag.py
View File

File diff suppressed because it is too large Load Diff

102
ipaserver/plugins/selfsign.py
View File

@@ -1,5 +1,6 @@
# Authors:
# Rob Crittenden <rcritten@@redhat.com>
# John Dennis <jdennis@redhat.com>
#
# Copyright (C) 2009 Red Hat
# see file 'COPYING' for use and warranty information
@@ -43,6 +44,7 @@ from ipaserver.plugins import rabase
from ipaserver.install import certs
import tempfile
from pyasn1 import error
from ipalib.request import ugettext as _
class ra(rabase.rabase):
"""
@@ -51,13 +53,32 @@ class ra(rabase.rabase):
def request_certificate(self, csr, request_type='pkcs10'):
"""
Submit certificate signing request.
:param csr: The certificate signing request.
:param request_type: The request type (defaults to ``'pkcs10'``).
"""
(csr_fd, csr_name) = tempfile.mkstemp()
Submit certificate signing request.
The command returns a dict with these possible key/value pairs.
Some key/value pairs may be absent.
+---------------+---------------+---------------+
|result name |result type |comments |
+===============+===============+===============+
|serial_number |unicode [1]_ | |
+---------------+---------------+---------------+
|certificate |unicode [2]_ | |
+---------------+---------------+---------------+
|request_id |unicode | |
+---------------+---------------+---------------+
|subject |unicode | |
+---------------+---------------+---------------+
.. [1] Passed through XMLRPC as decimal string. Can convert to
optimal integer type (int or long) via int(serial_number)
.. [2] Base64 encoded
"""
# certutil wants the CSR to have have a header and footer. Add one
# if it isn't there.
s = csr.find('-----BEGIN NEW CERTIFICATE REQUEST-----')
@@ -66,12 +87,47 @@ class ra(rabase.rabase):
if s == -1:
csr = '-----BEGIN NEW CERTIFICATE REQUEST-----\n' + csr + \
'-----END NEW CERTIFICATE REQUEST-----\n'
os.write(csr_fd, csr)
os.close(csr_fd)
(cert_fd, cert_name) = tempfile.mkstemp()
os.close(cert_fd)
serialno = certs.next_serial(self.serial_file)
try:
(csr_fd, csr_name) = tempfile.mkstemp()
os.write(csr_fd, csr)
os.close(csr_fd)
except Exception, e:
try:
os.remove(csr_name)
except:
pass
self.log.error('unable to create temporary csr file: %s' % e)
raise errors.CertificateOperationError(error=_('file operation'))
try:
(cert_fd, cert_name) = tempfile.mkstemp()
os.close(cert_fd)
except Exception, e:
try:
os.remove(csr_name)
except:
pass
try:
os.remove(cert_name)
except:
pass
self.log.error('unable to create temporary certificate file: %s' % e)
raise errors.CertificateOperationError(error=_('file operation'))
try:
serialno = certs.next_serial(self.serial_file)
except Exception, e:
try:
os.remove(csr_name)
except:
pass
try:
os.remove(cert_name)
except:
pass
self.log.error('next_serial() failed: %s' % e)
raise errors.CertificateOperationError(error=_('cannot obtain next serial number'))
try:
args = [
@@ -97,17 +153,31 @@ class ra(rabase.rabase):
p.stdin.write("0\n9\nn\n")
p.stdin.write("1\n9\nn\n")
(stdout, stderr) = p.communicate()
status = p.returncode
self.log.debug("stdout = %s" % stdout)
self.log.debug("stderr = %s" % stderr)
if status != 0:
try:
os.remove(cert_name)
except:
pass
self.log.error('certutil failed: %s' % stderr)
raise errors.CertificateOperationError(error=_('certutil failure'))
finally:
os.remove(csr_name)
try:
os.remove(csr_name)
except:
pass
try:
cert_fd = open(cert_name)
cert = cert_fd.read()
cert_fd.close()
finally:
os.remove(cert_name)
try:
os.remove(cert_name)
except:
pass
try:
# Grab the subject, reverse it, combine it and return it
@@ -120,7 +190,8 @@ class ra(rabase.rabase):
serial = x509.get_serial_number(cert)
except error.PyAsn1Error, e:
raise errors.GenericError(format='Unable to decode certificate in entry: %s' % str(e))
self.log.error('Unable to decode certificate in entry: %s' % str(e))
raise errors.CertificateOperationError(error='Unable to decode certificate in entry: %s' % str(e))
# To make it look like dogtag return just the base64 data.
cert = cert.replace('\n','')
@@ -130,6 +201,11 @@ class ra(rabase.rabase):
s = s + 27
cert = cert[s:e]
return {'status':0, 'subject': subject, 'certificate':cert, 'serial_number': "0x%x" % serial}
cmd_result = {}
cmd_result['serial_number'] = unicode(serial) # convert long to decimal unicode string
cmd_result['certificate'] = unicode(cert)
cmd_result['subject'] = unicode(subject)
return cmd_result
api.register(ra)