cert: allow search by certificate

Allow search by certificate data or file in cert-find.

https://fedorahosted.org/freeipa/ticket/5381

Reviewed-By: David Kupka <dkupka@redhat.com>
Reviewed-By: Pavel Vomacka <pvomacka@redhat.com>

4 files changed, 67 insertions, 11 deletions

diff --git a/API.txt b/API.txt
index ba942799c..142ffcd40 100644
--- a/API.txt
+++ b/API.txt
@@ -723,10 +723,11 @@ output: Entry('result')
 output: Output('summary', type=[<type 'unicode'>, <type 'NoneType'>])
 output: PrimaryKey('value')
 command: cert_find
-args: 1,28,4
+args: 1,29,4
 arg: Str('criteria?')
 option: Flag('all', autofill=True, cli_name='all', default=False)
 option: Str('cacn?', cli_name='ca')
+option: Bytes('certificate?', autofill=False)
 option: Flag('exactly?', autofill=True, default=False)
 option: Str('host*', cli_name='hosts')
 option: DateTime('issuedon_from?', autofill=False)
diff --git a/VERSION b/VERSION
index 46e17b95f..fe7614edd 100644
--- a/VERSION
+++ b/VERSION
@@ -90,5 +90,5 @@ IPA_DATA_VERSION=20100614120000
 #                                                      #
 ########################################################
 IPA_API_VERSION_MAJOR=2
-IPA_API_VERSION_MINOR=195
-# Last change: cert: add owner information
+IPA_API_VERSION_MINOR=196
+# Last change: cert: allow search by certificate
diff --git a/ipaclient/plugins/cert.py b/ipaclient/plugins/cert.py
index 7e8e15662..de4318b68 100644
--- a/ipaclient/plugins/cert.py
+++ b/ipaclient/plugins/cert.py
@@ -25,6 +25,7 @@ from ipalib import x509
 from ipalib import util
 from ipalib.parameters import File
 from ipalib.plugable import Registry
+from ipalib.text import _
 
 register = Registry()
 
@@ -51,3 +52,25 @@ class cert_show(MethodOverride):
                 raise errors.NoCertificateError(entry=keys[-1])
         else:
             return super(cert_show, self).forward(*keys, **options)
+
+
+@register(override=True)
+class cert_find(MethodOverride):
+    takes_options = (
+        File(
+            'file?',
+            label=_("Input filename"),
+            doc=_('File to load the certificate from.'),
+            include='cli',
+        ),
+    )
+
+    def forward(self, *args, **options):
+        if self.api.env.context == 'cli':
+            if 'certificate' in options and 'file' in options:
+                raise errors.MutuallyExclusiveError(
+                    reason=_("cannot specify both raw certificate and file"))
+            if 'certificate' not in options and 'file' in options:
+                options['certificate'] = x509.strip_header(options.pop('file'))
+
+        return super(cert_find, self).forward(*args, **options)
diff --git a/ipaserver/plugins/cert.py b/ipaserver/plugins/cert.py
index 39144ddf3..564d582c7 100644
--- a/ipaserver/plugins/cert.py
+++ b/ipaserver/plugins/cert.py
@@ -114,6 +114,12 @@ EXAMPLES:
  Search for certificates owned by a specific user:
    ipa cert-find --user=user
 
+ Examine a certificate:
+   ipa cert-find --file=cert.pem --all
+
+ Verify that a certificate is owner by a specific user:
+   ipa cert-find --file=cert.pem --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.
@@ -239,12 +245,17 @@ def caacl_check(principal_type, principal_string, ca, profile_id):
         )
 
 
+def validate_certificate(value):
+    return x509.validate_certificate(value, x509.DER)
+
+
 class BaseCertObject(Object):
     takes_params = (
         Bytes(
-            'certificate',
+            'certificate', validate_certificate,
             label=_("Certificate"),
             doc=_("Base-64 encoded certificate."),
+            normalizer=x509.normalize_certificate,
             flags={'no_create', 'no_update', 'no_search'},
         ),
         DNParam(
@@ -652,7 +663,7 @@ class cert(BaseCertObject):
         for param in super(cert, self).get_params():
             if param.name == 'serial_number':
                 param = param.clone(primary_key=True)
-            elif param.name == 'issuer':
+            elif param.name in ('certificate', 'issuer'):
                 param = param.clone(flags=param.flags - {'no_search'})
             yield param
 
@@ -978,6 +989,7 @@ class cert_find(Search, CertMethod):
             any(name in options for name in ca_options - {'exactly'}) or
             options['exactly'])
         has_ldap_options = any(name in options for name in ldap_options)
+        has_cert_option = 'certificate' in options
 
         try:
             ca_enabled_check()
@@ -1006,6 +1018,12 @@ class cert_find(Search, CertMethod):
         obj_dict = {}
         truncated = False
 
+        if has_cert_option:
+            cert = options['certificate']
+            obj = {'certificate': unicode(base64.b64encode(cert))}
+            obj_seq.append(obj)
+            obj_dict[cert] = obj
+
         if ca_enabled:
             ra_options = {}
             for name, value in options.items():
@@ -1025,23 +1043,37 @@ class cert_find(Search, CertMethod):
                 if ((not pkey_only and all) or
                         not no_members or
                         not has_ca_options or
-                        has_ldap_options):
+                        has_ldap_options or
+                        has_cert_option):
                     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)
+                    try:
+                        obj = obj_dict[cert]
+                    except KeyError:
+                        if has_cert_option:
+                            continue
+                        obj = {}
+                        obj_seq.append(obj)
+                        obj_dict[cert] = obj
+                else:
+                    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):
+                has_ldap_options or
+                has_cert_option):
             ldap = self.api.Backend.ldap2
 
             filters = []
-            cert_filter = '(usercertificate=*)'
+            if 'certificate' in options:
+                cert_filter = ldap.make_filter_from_attr(
+                    'usercertificate', options['certificate'])
+            else:
+                cert_filter = '(usercertificate=*)'
             filters.append(cert_filter)
             for owner in self.obj._owners():
                 oc_filter = ldap.make_filter_from_attr(
@@ -1077,7 +1109,7 @@ class cert_find(Search, CertMethod):
                         try:
                             obj = obj_dict[cert]
                         except KeyError:
-                            if has_ca_options:
+                            if has_ca_options or has_cert_option:
                                 continue
                             obj = {
                                 'certificate': unicode(base64.b64encode(cert))}