On 14.6.2016 11:44, Jan Cholasta wrote:
On 21.4.2016 09:11, Jan Cholasta wrote:
On 6.4.2016 15:46, Pavel Vomacka wrote:
On 03/16/2016 01:50 PM, Jan Cholasta wrote:
Hi,
the attached patches implement the server-side part of
<https://fedorahosted.org/freeipa/ticket/5381>.
Honza
Hi,
thank you for the patches. I tested them and they work well. But I would
like to ask you whether would be possible to extend the response of
'basecert_find' method and probably also 'basecert_show' response. I
think of these information:
1) information whether the certificate is issued by our CA or not.
You can check for that by comparing the issuer name of the certificate
to "CN=Certificate Authority,$SUBJECT_BASE". You can get subject base
from config-show.
2) this probably wouldn't be possible (as we discussed), but I rather
write it too - the information about revocation reason. The same as the
'cert_show' provides.
Added --check-revocation flag to request this information. Currently it
works only on certificates issued by our CA.
3) MD5 and SHA1 fingerprints as the 'cert_show' method returns
Added, also included SHA-256.
Thank you again.
Updated patches attached.
Updated and rebased patches attached. Requires Fraser's sub-CA patches.
Attaching updated patch 623, which fixes these issues found by David:
<https://paste.fedoraproject.org/378997/65913663/>.
--
Jan Cholasta
From 4afb5a1bee3e2d152052d6ed924cafe4c1c51187 Mon Sep 17 00:00:00 2001
From: Jan Cholasta <jchol...@redhat.com>
Date: Tue, 14 Jun 2016 09:09:10 +0200
Subject: [PATCH 3/4] cert: add owner information
Get owner information from LDAP in cert-show and cert-find. Allow search by
owner in cert-find.
https://fedorahosted.org/freeipa/ticket/5381
---
API.txt | 15 ++-
VERSION | 4 +-
ipaserver/plugins/cert.py | 267 ++++++++++++++++++++++++++++++++++++++++------
3 files changed, 251 insertions(+), 35 deletions(-)
diff --git a/API.txt b/API.txt
index f01d3c1..94bf337 100644
--- a/API.txt
+++ b/API.txt
@@ -730,23 +730,33 @@ output: Entry('result')
output: Output('summary', type=[<type 'unicode'>, <type 'NoneType'>])
output: PrimaryKey('value')
command: cert_find
-args: 1,20,4
+args: 1,30,4
arg: Str('criteria?')
option: Flag('all', autofill=True, cli_name='all', default=False)
option: Str('cacn?', cli_name='ca')
option: Flag('exactly?', autofill=True, default=False)
+option: Str('host*', cli_name='hosts')
+option: Str('idoverrideuser*', cli_name='idoverrideusers')
option: DateTime('issuedon_from?', autofill=False)
option: DateTime('issuedon_to?', autofill=False)
option: DNParam('issuer?', autofill=False)
option: Int('max_serial_number?', autofill=False)
option: Int('min_serial_number?', autofill=False)
+option: Str('no_host*', cli_name='no_hosts')
+option: Str('no_idoverrideuser*', cli_name='no_idoverrideusers')
+option: Flag('no_members', autofill=True, default=True)
+option: Str('no_service*', cli_name='no_services')
+option: Str('no_user*', cli_name='no_users')
option: Flag('pkey_only?', autofill=True, default=False)
option: Flag('raw', autofill=True, cli_name='raw', default=False)
option: Int('revocation_reason?', autofill=False, default=0)
option: DateTime('revokedon_from?', autofill=False)
option: DateTime('revokedon_to?', autofill=False)
+option: Str('service*', cli_name='services')
option: Int('sizelimit?')
option: Str('subject?', autofill=False)
+option: Int('timelimit?')
+option: Str('user*', cli_name='users')
option: DateTime('validnotafter_from?', autofill=False)
option: DateTime('validnotafter_to?', autofill=False)
option: DateTime('validnotbefore_from?', autofill=False)
@@ -782,10 +792,11 @@ option: Int('revocation_reason', autofill=True, default=0)
option: Str('version?')
output: Output('result')
command: cert_show
-args: 1,5,3
+args: 1,6,3
arg: Int('serial_number')
option: Flag('all', autofill=True, cli_name='all', default=False)
option: Str('cacn?', cli_name='ca')
+option: Flag('no_members', autofill=True, default=False)
option: Str('out?')
option: Flag('raw', autofill=True, cli_name='raw', default=False)
option: Str('version?')
diff --git a/VERSION b/VERSION
index 354aa62..a383578 100644
--- a/VERSION
+++ b/VERSION
@@ -90,5 +90,5 @@ IPA_DATA_VERSION=20100614120000
# #
########################################################
IPA_API_VERSION_MAJOR=2
-IPA_API_VERSION_MINOR=185
-# Last change: cert: add object plugin
+IPA_API_VERSION_MINOR=186
+# Last change: cert: add owner information
diff --git a/ipaserver/plugins/cert.py b/ipaserver/plugins/cert.py
index 9364fdd..5698c3a 100644
--- a/ipaserver/plugins/cert.py
+++ b/ipaserver/plugins/cert.py
@@ -19,6 +19,7 @@
# 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 binascii
import datetime
import os
@@ -110,6 +111,9 @@ EXAMPLES:
Search for certificates based on issuance date
ipa cert-find --issuedon-from=2013-02-01 --issuedon-to=2013-02-07
+ Search for certificates owned by a specific user:
+ ipa cert-find --user=user
+
IPA currently immediately issues (or declines) all certificate requests so
the status of a request is not normally useful. This is for future use
or the case where a CA does not immediately issue a certificate.
@@ -656,9 +660,48 @@ class cert(BaseCertObject):
param = param.clone(flags=param.flags - {'no_search'})
yield param
+ for owner in self._owners():
+ yield owner.primary_key.clone_rename(
+ 'owner_{0}'.format(owner.name),
+ required=False,
+ multivalue=True,
+ primary_key=False,
+ label=_("Owner %s") % owner.object_name,
+ flags={'no_create', 'no_update', 'no_search'},
+ )
+
+ def _owners(self):
+ for name in ('user', 'host', 'service', 'idoverrideuser'):
+ yield self.api.Object[name]
+
+ def _fill_owners(self, obj):
+ for owner in self._owners():
+ container_dn = DN(owner.container_dn, self.api.env.basedn)
+ name = 'owner_' + owner.name
+ for dn in obj['owner']:
+ if dn.endswith(container_dn, 1):
+ value = owner.get_primary_key_from_dn(dn)
+ obj.setdefault(name, []).append(value)
+
+
+class CertMethod(BaseCertMethod):
+ def get_options(self):
+ for option in super(CertMethod, self).get_options():
+ yield option
+
+ for o in self.has_output:
+ if isinstance(o, (output.Entry, output.ListOfEntries)):
+ yield Flag(
+ 'no_members',
+ doc=_("Suppress processing of membership attributes."),
+ exclude='webui',
+ flags={'no_output'},
+ )
+ break
+
@register()
-class cert_show(Retrieve, BaseCertMethod, VirtualCommand):
+class cert_show(Retrieve, CertMethod, VirtualCommand):
__doc__ = _('Retrieve an existing certificate.')
takes_options = (
@@ -671,7 +714,8 @@ class cert_show(Retrieve, BaseCertMethod, VirtualCommand):
operation="retrieve certificate"
- def execute(self, serial_number, all=False, raw=False, **options):
+ def execute(self, serial_number, all=False, raw=False, no_members=False,
+ **options):
ca_enabled_check()
hostname = None
try:
@@ -701,10 +745,26 @@ class cert_show(Retrieve, BaseCertMethod, VirtualCommand):
"issued by CA '%(ca)s' not found")
% dict(serial=serial_number, ca=options['cacn']))
+ if all or not no_members:
+ ldap = self.api.Backend.ldap2
+ filter = ldap.make_filter_from_attr(
+ 'usercertificate', base64.b64decode(result['certificate']))
+ try:
+ entries = ldap.get_entries(base_dn=self.api.env.basedn,
+ filter=filter,
+ attrs_list=[''])
+ except errors.EmptyResult:
+ entries = []
+ for entry in entries:
+ result.setdefault('owner', []).append(entry.dn)
+
if not raw:
result['certificate'] = result['certificate'].replace('\r\n', '')
self.obj._parse(result)
result['revoked'] = ('revocation_reason' in result)
+ if 'owner' in result:
+ self.obj._fill_owners(result)
+ del result['owner']
if hostname:
# If we have a hostname we want to verify that the subject
@@ -716,7 +776,7 @@ class cert_show(Retrieve, BaseCertMethod, VirtualCommand):
@register()
-class cert_revoke(PKQuery, BaseCertMethod, VirtualCommand):
+class cert_revoke(PKQuery, CertMethod, VirtualCommand):
__doc__ = _('Revoke a certificate.')
operation = "revoke certificate"
@@ -752,7 +812,7 @@ class cert_revoke(PKQuery, BaseCertMethod, VirtualCommand):
@register()
-class cert_remove_hold(PKQuery, BaseCertMethod, VirtualCommand):
+class cert_remove_hold(PKQuery, CertMethod, VirtualCommand):
__doc__ = _('Take a revoked certificate off hold.')
has_output_params = (
@@ -781,7 +841,7 @@ class cert_remove_hold(PKQuery, BaseCertMethod, VirtualCommand):
@register()
-class cert_find(Search, BaseCertMethod):
+class cert_find(Search, CertMethod):
__doc__ = _('Search for existing certificates.')
takes_options = (
@@ -851,6 +911,11 @@ class cert_find(Search, BaseCertMethod):
doc=_("Results should contain primary key attribute only "
"(\"certificate\")"),
),
+ Int('timelimit?',
+ label=_('Time Limit'),
+ doc=_('Time limit of search in seconds (0 is unlimited)'),
+ minvalue=0,
+ ),
Int('sizelimit?',
label=_("Size Limit"),
doc=_("Maximum number of entries returned (0 is unlimited)"),
@@ -862,9 +927,63 @@ class cert_find(Search, BaseCertMethod):
'%(count)d certificate matched', '%(count)d certificates matched', 0
)
+ def get_options(self):
+ for option in super(cert_find, self).get_options():
+ if option.name == 'no_members':
+ option = option.clone(default=True,
+ flags=set(option.flags) | {'no_option'})
+ yield option
+
+ for owner in self.obj._owners():
+ yield owner.primary_key.clone_rename(
+ '{0}'.format(owner.name),
+ required=False,
+ multivalue=True,
+ primary_key=False,
+ query=True,
+ cli_name='{0}s'.format(owner.name),
+ doc=(_("Search for certificates with these owner %s.") %
+ owner.object_name_plural),
+ label=owner.object_name,
+ )
+ yield owner.primary_key.clone_rename(
+ 'no_{0}'.format(owner.name),
+ required=False,
+ multivalue=True,
+ primary_key=False,
+ query=True,
+ cli_name='no_{0}s'.format(owner.name),
+ doc=(_("Search for certificates without these owner %s.") %
+ owner.object_name_plural),
+ label=owner.object_name,
+ )
+
def execute(self, criteria=None, all=False, raw=False, pkey_only=False,
- sizelimit=None, **options):
- ca_enabled_check()
+ no_members=True, timelimit=None, sizelimit=None, **options):
+ ca_options = {'cacn',
+ 'revocation_reason',
+ 'issuer',
+ 'subject',
+ 'min_serial_number', 'max_serial_number',
+ 'exactly',
+ 'validnotafter_from', 'validnotafter_to',
+ 'validnotbefore_from', 'validnotbefore_to',
+ 'issuedon_from', 'issuedon_to',
+ 'revokedon_from', 'revokedon_to'}
+ ldap_options = {prefix + owner.name
+ for owner in self.obj._owners()
+ for prefix in ('', 'no_')}
+ has_ca_options = any(options.get(name) for name in ca_options)
+ has_ldap_options = any(options.get(name) for name in ldap_options)
+
+ try:
+ ca_enabled_check()
+ except errors.NotFound:
+ if has_ca_options:
+ raise
+ ca_enabled = False
+ else:
+ ca_enabled = True
if 'cacn' in options:
ca_obj = api.Command.ca_show(options['cacn'])['result']
@@ -881,47 +1000,133 @@ class cert_find(Search, BaseCertMethod):
return dict(result=[], count=0, truncated=False)
obj_seq = []
+ obj_dict = {}
+ truncated = False
+
+ if ca_enabled:
+ ra_options = {}
+ for name, value in options.items():
+ if name not in ca_options:
+ continue
+ if isinstance(value, datetime.datetime):
+ value = value.strftime(PKIDATE_FORMAT)
+ ra_options[name] = value
+ if sizelimit is not None:
+ ra_options['sizelimit'] = sizelimit
+
+ for ra_obj in self.Backend.ra.find(ra_options):
+ obj = {}
+ if ((not pkey_only and all) or
+ not no_members or
+ not has_ca_options or
+ has_ldap_options):
+ ra_obj.update(
+ self.Backend.ra.get_certificate(
+ str(ra_obj['serial_number'])))
+ cert = base64.b64decode(ra_obj['certificate'])
+ obj_dict[cert] = obj
+ obj_seq.append(obj)
+ obj.update(ra_obj)
+
+ if ((not pkey_only and all) or
+ not no_members or
+ not has_ca_options or
+ has_ldap_options):
+ ldap = self.api.Backend.ldap2
+
+ filters = []
+ cert_filter = '(usercertificate=*)'
+ filters.append(cert_filter)
+ for owner in self.obj._owners():
+ oc_filter = ldap.make_filter_from_attr(
+ 'objectclass', owner.object_class, ldap.MATCH_ALL)
+ for prefix, rule in (('', ldap.MATCH_ALL),
+ ('no_', ldap.MATCH_NONE)):
+ value = options.get(prefix + owner.name)
+ if value is None:
+ continue
+ pkey_filter = ldap.make_filter_from_attr(
+ owner.primary_key.name, value, rule)
+ filters.append(oc_filter)
+ filters.append(pkey_filter)
+ filter = ldap.combine_filters(filters, ldap.MATCH_ALL)
- ra_options = {}
- for name, value in options.items():
- if isinstance(value, datetime.datetime):
- value = value.strftime(PKIDATE_FORMAT)
- ra_options[name] = value
- if sizelimit is not None:
- ra_options['sizelimit'] = sizelimit
-
- for ra_obj in self.Backend.ra.find(ra_options):
- obj = {}
- if all:
- ra_obj.update(
- self.Backend.ra.get_certificate(
- str(ra_obj['serial_number'])))
- obj_seq.append(obj)
- obj.update(ra_obj)
+ try:
+ entries, truncated = ldap.find_entries(
+ base_dn=self.api.env.basedn,
+ filter=filter,
+ attrs_list=['usercertificate'],
+ time_limit=timelimit,
+ size_limit=sizelimit,
+ )
+ except errors.EmptyResult:
+ entries, truncated = [], False
+ for entry in entries:
+ seen = set()
+ for attr in ('usercertificate', 'usercertificate;binary'):
+ for cert in entry.get(attr, []):
+ if cert in seen:
+ continue
+ seen.add(cert)
+ try:
+ obj = obj_dict[cert]
+ except KeyError:
+ if has_ca_options:
+ continue
+ obj = {
+ 'certificate': unicode(base64.b64encode(cert))}
+ obj_seq.append(obj)
+ obj_dict[cert] = obj
+ obj.setdefault('owner', []).append(entry.dn)
result = []
for obj in obj_seq:
+ if has_ldap_options and 'owner' not in obj:
+ continue
if not pkey_only:
if not raw:
if 'certificate' in obj:
obj['certificate'] = (
obj['certificate'].replace('\r\n', ''))
self.obj._parse(obj)
- obj['subject'] = DN(obj['subject'])
- obj['issuer'] = DN(obj['issuer'])
- obj['revoked'] = (
- obj['status'] in (u'REVOKED', u'REVOKED_EXPIRED'))
+ if not all:
+ del obj['certificate']
+ del obj['valid_not_before']
+ del obj['valid_not_after']
+ del obj['md5_fingerprint']
+ del obj['sha1_fingerprint']
+ if 'subject' in obj:
+ obj['subject'] = DN(obj['subject'])
+ if 'issuer' in obj:
+ obj['issuer'] = DN(obj['issuer'])
+ if 'status' in obj:
+ obj['revoked'] = (
+ obj['status'] in (u'REVOKED', u'REVOKED_EXPIRED'))
+ if 'owner' in obj:
+ if all or not no_members:
+ self.obj._fill_owners(obj)
+ del obj['owner']
+ else:
+ if 'certificate' in obj:
+ if not all:
+ del obj['certificate']
+ if 'owner' in obj:
+ if not all and no_members:
+ del obj['owner']
else:
- serial_number = obj['serial_number']
- obj.clear()
- obj['serial_number'] = serial_number
+ if 'serial_number' in obj:
+ serial_number = obj['serial_number']
+ obj.clear()
+ obj['serial_number'] = serial_number
+ else:
+ obj.clear()
result.append(obj)
ret = dict(
result=result
)
ret['count'] = len(ret['result'])
- ret['truncated'] = False
+ ret['truncated'] = bool(truncated)
return ret
--
2.7.4
--
Manage your subscription for the Freeipa-devel mailing list:
https://www.redhat.com/mailman/listinfo/freeipa-devel
Contribute to FreeIPA: http://www.freeipa.org/page/Contribute/Code
