1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
|
# Authors:
# Andrew Wnuk <awnuk@redhat.com>
# Jason Gerard DeRose <jderose@redhat.com>
#
# Copyright (C) 2009 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
"""
Command plugins for IPA-RA certificate operations.
"""
from ipalib import api, SkipPluginModule
if api.env.enable_ra is not True:
# In this case, abort loading this plugin module...
raise SkipPluginModule(reason='env.enable_ra is not True')
from ipalib import Command, Str, Int, Bytes, Flag
from ipalib import errors
from ipalib.plugins.virtual import *
import base64
from OpenSSL import crypto
def get_serial(certificate):
"""
Given a certificate, return the serial number in that cert
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.
"""
if type(certificate) in (list, tuple):
certificate = certificate[0]
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')
return serial
def validate_csr(ugettext, csr):
"""
For now just verify that it is properly base64-encoded.
"""
try:
base64.b64decode(csr)
except Exception, e:
raise errors.Base64DecodeError(reason=str(e))
class cert_request(VirtualCommand):
"""
Submit a certificate singing request.
"""
takes_args = (Str('csr', validate_csr),)
operation="request certificate"
takes_options = (
Str('principal',
doc="service principal for this certificate (e.g. HTTP/test.example.com)",
),
Str('request_type',
default=u'pkcs10',
autofill=True,
),
Flag('add',
doc="automatically add the principal if it doesn't exist",
default=False,
autofill=True
),
)
def execute(self, csr, **kw):
super(cert_request, self).execute()
skw = {"all": True}
principal = kw.get('principal')
add = kw.get('add')
del kw['principal']
del kw['add']
service = None
# See if the service exists and punt if it doesn't and we aren't
# going to add it
try:
(dn, service) = api.Command['service_show'](principal, **skw)
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(service['usercertificate']))
except errors.NotFound, e:
if not add:
raise errors.NotFound(reason="The service principal for this request doesn't exist.")
# Request the certificate
result = self.Backend.ra.request_certificate(csr, **kw)
# Success? Then add it to the service entry. We know that it
# either exists or we should add it.
if result.get('status') == '0':
if service is None:
service = api.Command['service_add'](principal, **{})
skw = {"usercertificate": str(result.get('certificate'))}
api.Command['service_mod'](principal, **skw)
return result
def output_for_cli(self, textui, result, *args, **kw):
if isinstance(result, dict) and len(result) > 0:
textui.print_entry(result, 0)
else:
textui.print_plain('Failed to submit a certificate request.')
def run(self, *args, **options):
"""
Dispatch to forward() and execute() to do work locally and on the
server.
"""
if self.env.in_server:
return self.execute(*args, **options)
# Client-side code
csr = args[0]
if csr[:7] == "file://":
file = csr[7:]
try:
f = open(file, "r")
csr = f.readlines()
f.close()
except IOError, err:
raise errors.ValidationError(name='csr', error=err[1])
csr = "".join(csr)
# We just want the CSR bits, make sure there is nothing else
s = csr.find("-----BEGIN NEW CERTIFICATE REQUEST-----")
e = csr.find("-----END NEW CERTIFICATE REQUEST-----")
if s >= 0:
csr = csr[s+40:e]
csr = csr.decode('UTF-8')
return self.forward(csr, **options)
api.register(cert_request)
class cert_status(VirtualCommand):
"""
Check status of a certificate signing request.
"""
takes_args = ('request_id')
operation = "certificate status"
def execute(self, request_id, **kw):
super(cert_status, self).execute()
return self.Backend.ra.check_request_status(request_id)
def output_for_cli(self, textui, result, *args, **kw):
if isinstance(result, dict) and len(result) > 0:
textui.print_entry(result, 0)
else:
textui.print_plain('Failed to retrieve a request status.')
api.register(cert_status)
class cert_get(VirtualCommand):
"""
Retrieve an existing certificate.
"""
takes_args = ('serial_number')
operation="retrieve certificate"
def execute(self, serial_number):
super(cert_get, self).execute()
return self.Backend.ra.get_certificate(serial_number)
def output_for_cli(self, textui, result, *args, **kw):
if isinstance(result, dict) and len(result) > 0:
textui.print_entry(result, 0)
else:
textui.print_plain('Failed to obtain a certificate.')
api.register(cert_get)
class cert_revoke(VirtualCommand):
"""
Revoke a certificate.
"""
takes_args = ('serial_number')
operation = "revoke certificate"
# FIXME: The default is 0. Is this really an Int param?
takes_options = (
Int('revocation_reason?',
doc='Reason for revoking the certificate (0-10)',
minvalue=0,
maxvalue=10,
default=0,
),
)
def execute(self, serial_number, **kw):
super(cert_revoke, self).execute()
return self.Backend.ra.revoke_certificate(serial_number, **kw)
def output_for_cli(self, textui, result, *args, **kw):
if isinstance(result, dict) and len(result) > 0:
textui.print_entry(result, 0)
else:
textui.print_plain('Failed to revoke a certificate.')
api.register(cert_revoke)
class cert_remove_hold(VirtualCommand):
"""
Take a revoked certificate off hold.
"""
takes_args = ('serial_number')
operation = "certificate remove hold"
def execute(self, serial_number, **kw):
super(cert_remove_hold, self).execute()
return self.Backend.ra.take_certificate_off_hold(serial_number)
def output_for_cli(self, textui, result, *args, **kw):
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.')
api.register(cert_remove_hold)
|
