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
|
# 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
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:
service = api.Command['service_show'](principal, **skw)
if service.get('usercertificate'):
# FIXME, what to do here? Do we revoke the old cert?
raise errors.GenericError(format='entry already has a certificate')
except errors.NotFound, e:
if not add:
raise e
# 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.')
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)
|
