On 07/02/2015 02:37 PM, Martin Babinsky wrote:
On 07/02/2015 11:28 AM, Martin Babinsky wrote:
On 07/02/2015 11:12 AM, Martin Babinsky wrote:
On 07/01/2015 03:05 PM, Martin Babinsky wrote:
On 06/30/2015 02:45 PM, Martin Babinsky wrote:
On 06/30/2015 01:11 PM, Martin Babinsky wrote:
On 06/30/2015 12:04 PM, Jan Cholasta wrote:
Dne 29.6.2015 v 10:36 Martin Babinsky napsal(a):
On 06/23/2015 01:49 PM, Martin Babinsky wrote:
This patchset implements new API commands for manipulating
user/host/service userCertificate attribute alongside some
underlying
plumbing.
PATCH 0045 is a small test suite that I slapped together since
manual
testing of this stuff is very cumbersome. It requires my PATCH
0040 to
apply and work which was pushed to master recently
(commit 74883bbc959058c8bfafd9f63e8fad0e3792ac28).
The work is related to
http://www.freeipa.org/page/V4/User_Certificates
and https://fedorahosted.org/freeipa/ticket/4238
Attaching updated patches.
Here are some notes for Jan because I did some things differently
than
we agreed on during review:
1.) I chose not to rename 'usercertificate' to
'usercertificate;binary'
and back in pre/post callbacks. Despite the fact that the correct
way to
name the certificate attribute is 'usercertificate;binary', I feel
that
suddenly renaming it in the new code is asking for trouble.
New code is new, there is no renaming, there is naming, and that
naming
should follow standards, and the standard is userCertificate;binary.
(For the record I did not ask for any renaming in *old* host and
service
code.)
OK I will then use 'usercertificate;binary' and try to not break
things.
I'm all for changing the mapping between CLI options and actual
attribute names but it should be done in a systematic fashion.
+1, shall I post a patch?
That would be great, but I'm not sure if there is time for it.
Maybe we
can create a ticket for tracking?
2.) I have kept the `normalize_certs` function. It has the
potential to
catch incorrectly formatted/encoded certificates and in a way
circumvents the slightly demented way the framework deals with
supposedly binary data.
One sentence above you asked for doing things in systematic fashion.
This is exactly what it isn't. A systematic solution would be a new
parameter type for certificates.
Ha I didn't notice that incorrect encoding is caught by validator.
But I think that we still need to catch malformed certificates that
can
not be decoded to DER and AFAIK we don't do that anywhere (failing
tests
when adding a random Base64-encoded string confirm this).
All this probably stems from my confusion about the way IPA framework
guesses binary data. For example, if I call
`api.Command.user_add_cert`
and fill 'certificate' option with Base64 blob reencoded to Unicode,
everything works as expected.
However, filling this option with 'str' leads to another round of
Base64
encoding in the framework, leading to 'userCertificate;binary'
which is
filled by original Base64 blob instead of DER encoded cert.
I have also added two negative test cases which deal with
incorrectly
encoded and formatted certificates.
Attaching updated patches (actually only 44 is updated, I added the
rename to/from 'usercertificate;binary' to user pre/post callbacks).
Another patch update attached (mainly fixing pep8 complaints and
reworking certificate validation).
Updated patches attached.
I left a a bug in PATCH 0043. Attaching updated version.
Attaching updated patches.
Attaching revised patchset.
It would be nice if Milan could comment on PATCH 0045.
--
Martin^3 Babinsky
From 5692c7759a4c9249edd59ea68745d20679d03419 Mon Sep 17 00:00:00 2001
From: Martin Babinsky <mbabi...@redhat.com>
Date: Tue, 23 Jun 2015 13:42:45 +0200
Subject: [PATCH 4/4] test suite for user/host/service certificate management
API commands
These tests excercise various scenarios when using new class of API commands
to add or remove certificates to user/service/host entries.
Part of http://www.freeipa.org/page/V4/User_Certificates
---
ipatests/test_xmlrpc/test_add_remove_cert_cmd.py | 349 +++++++++++++++++++++++
1 file changed, 349 insertions(+)
create mode 100644 ipatests/test_xmlrpc/test_add_remove_cert_cmd.py
diff --git a/ipatests/test_xmlrpc/test_add_remove_cert_cmd.py b/ipatests/test_xmlrpc/test_add_remove_cert_cmd.py
new file mode 100644
index 0000000000000000000000000000000000000000..c414498a29ba2ffaf0c6fb0d2a9e92a34a673b70
--- /dev/null
+++ b/ipatests/test_xmlrpc/test_add_remove_cert_cmd.py
@@ -0,0 +1,349 @@
+#
+# Copyright (C) 2015 FreeIPA Contributors see COPYING for license
+#
+
+import base64
+
+from ipalib import api, errors
+
+from ipatests.util import assert_deepequal, raises
+from xmlrpc_test import XMLRPC_test
+from ipapython.dn import DN
+from testcert import get_testcert
+
+
+class CertManipCmdTestBase(XMLRPC_test):
+ entity_class = ''
+ entity_pkey = None
+ entity_subject = None
+ entity_principal = None
+ non_existent_entity = None
+
+ profile_store_orig = True
+ default_profile_id = u'caIPAserviceCert'
+ default_caacl = u'hosts_services_%s' % default_profile_id
+ cmd_options = dict(
+ entity_add=None,
+ caacl=None,
+ )
+ cert_add_cmd = None
+ cert_del_cmd = None
+
+ cert_add_summary = u''
+ cert_del_summary = u''
+
+ entity_attrs = None
+
+ @classmethod
+ def disable_profile_store(cls):
+ try:
+ api.Command.certprofile_mod(cls.default_profile_id,
+ ipacertprofilestoreissued=False)
+ except errors.EmptyModlist:
+ cls.profile_store_orig = False
+ else:
+ cls.profile_store_orig = True
+
+ @classmethod
+ def restore_profile_store(cls):
+ if cls.profile_store_orig:
+ api.Command.certprofile_mod(
+ cls.default_profile_id,
+ ipacertprofilestoreissued=cls.profile_store_orig)
+
+ @classmethod
+ def add_entity(cls):
+ api.Command['%s_add' % cls.entity_class](
+ cls.entity_pkey,
+ **cls.cmd_options['entity_add'])
+
+ @classmethod
+ def delete_entity(cls):
+ try:
+ api.Command['%s_del' % cls.entity_class](cls.entity_pkey)
+ except errors.NotFound:
+ pass
+
+ # optional methods which implement adding CA ACL rule so that we can
+ # request cert for the entity. Currently used only for users.
+ @classmethod
+ def add_caacl(cls):
+ pass
+
+ @classmethod
+ def remove_caacl(cls):
+ pass
+
+ @classmethod
+ def setup_class(cls):
+ super(CertManipCmdTestBase, cls).setup_class()
+
+ cls.delete_entity()
+
+ cls.add_entity()
+ cls.add_caacl()
+
+ cls.disable_profile_store()
+
+ # list of certificates to add to entry
+ cls.certs = [
+ get_testcert(DN(('CN', cls.entity_subject)), cls.entity_principal)
+ for i in xrange(3)
+ ]
+
+ # list of certificates for testing of removal of non-existent certs
+ cls.nonexistent_certs = [
+ get_testcert(DN(('CN', cls.entity_subject)), cls.entity_principal)
+ for j in xrange(2)
+ ]
+
+ # cert subset to remove from entry
+ cls.certs_subset = cls.certs[:2]
+
+ # remaining subset
+ cls.certs_remainder = cls.certs[2:]
+
+ # mixture of certs which exist and do not exists in the entry
+ cls.mixed_certs = cls.certs[:2] + cls.nonexistent_certs[:1]
+
+ # invalid base64 encoding
+ cls.invalid_b64 = [u'few4w24gvrae54y6463234f']
+
+ # malformed certificate
+ cls.malformed_cert = [base64.b64encode('malformed cert')]
+
+ # store entity info for the final test
+ cls.entity_attrs = api.Command['%s_show' % cls.entity_class](
+ cls.entity_pkey)
+
+ @classmethod
+ def teardown_class(cls):
+ cls.delete_entity()
+ cls.remove_caacl()
+
+ cls.restore_profile_store()
+ super(CertManipCmdTestBase, cls).teardown_class()
+
+ def add_certs(self, certs):
+ # pylint: disable=E1102
+ result = self.cert_add_cmd(self.entity_pkey, usercertificate=certs)
+ return dict(
+ usercertificate=result['result'].get('usercertificate', []),
+ value=result.get('value'),
+ summary=result.get('summary')
+ )
+
+ def remove_certs(self, certs):
+ # pylint: disable=E1102
+ result = self.cert_del_cmd(self.entity_pkey, usercertificate=certs)
+ return dict(
+ usercertificate=result['result'].get('usercertificate', []),
+ value=result.get('value'),
+ summary=result.get('summary')
+ )
+
+ def test_01_add_cert_to_nonexistent_entity(self):
+ """
+ Tests whether trying to add certificates to a non-existent entry
+ raises NotFound error.
+ """
+ raises(errors.NotFound, self.cert_add_cmd,
+ self.non_existent_entity, usercertificate=self.certs)
+
+ def test_02_remove_cert_from_nonexistent_entity(self):
+ """
+ Tests whether trying to remove certificates from a non-existent entry
+ raises NotFound error.
+ """
+ raises(errors.NotFound, self.cert_add_cmd,
+ self.non_existent_entity, usercertificate=self.certs)
+
+ def test_03_remove_cert_from_entity_with_no_certs(self):
+ """
+ Attempt to remove certificates from an entity that has none raises
+ AttrValueNotFound
+ """
+ raises(errors.AttrValueNotFound, self.remove_certs, self.certs)
+
+ def test_04_add_invalid_b64_blob_to_entity(self):
+ raises(errors.Base64DecodeError, self.add_certs, self.invalid_b64)
+
+ def test_05_add_malformed_cert_to_entity(self):
+ raises(errors.CertificateFormatError, self.add_certs,
+ self.malformed_cert)
+
+ def test_06_add_single_cert_to_entity(self):
+ """
+ Add single certificate to entry
+ """
+ assert_deepequal(
+ dict(
+ usercertificate=[base64.b64decode(self.certs[0])],
+ summary=self.cert_add_summary % self.entity_pkey,
+ value=self.entity_pkey,
+ ),
+ self.add_certs([self.certs[0]])
+ )
+
+ def test_07_add_more_certs_to_entity(self):
+ """
+ Add the rest of the certificate set to the entry.
+ """
+ assert_deepequal(
+ dict(
+ usercertificate=map(base64.b64decode, self.certs),
+ summary=self.cert_add_summary % self.entity_pkey,
+ value=self.entity_pkey,
+ ),
+ self.add_certs(self.certs[1:])
+ )
+
+ def test_08_add_already_present_cert_to_entity(self):
+ """
+ Tests that ExecutionError is raised when attempting to add certificates
+ to the entry that already contains them.
+ """
+ raises(
+ errors.ExecutionError,
+ self.add_certs,
+ self.certs_subset
+ )
+
+ def test_09_remove_nonexistent_certs_from_entity(self):
+ """
+ Tests that an attempt to remove certificates that are not present in
+ the entry raises AttrValueNotFound
+ """
+ raises(
+ errors.AttrValueNotFound,
+ self.remove_certs,
+ self.nonexistent_certs
+ )
+
+ def test_10_remove_valid_and_nonexistent_certs_from_entity(self):
+ """
+ Try to remove multiple certificates. Some of them are not present in
+ the entry. This scenario should raise InvocationError.
+ """
+ raises(
+ errors.AttrValueNotFound,
+ self.remove_certs,
+ self.mixed_certs
+ )
+
+ def test_11_remove_cert_subset_from_entity(self):
+ """
+ Test correct removal of a subset of entry's certificates.
+ """
+ assert_deepequal(
+ dict(
+ usercertificate=map(base64.b64decode,
+ self.certs_remainder),
+ summary=self.cert_del_summary % self.entity_pkey,
+ value=self.entity_pkey,
+ ),
+ self.remove_certs(self.certs_subset)
+ )
+
+ def test_12_remove_remaining_certs_from_entity(self):
+ """
+ Test correct removal of all the remaining certificates from the entry.
+ """
+ assert_deepequal(
+ dict(
+ usercertificate=[],
+ summary=self.cert_del_summary % self.entity_pkey,
+ value=self.entity_pkey,
+ ),
+ self.remove_certs(self.certs_remainder)
+ )
+
+ def test_99_check_final_entity_consistency(self):
+ """
+ Tests that all the previous operations do not modify other attributes
+ of the entry. Make sure that the show command returns the same
+ information as in the beginning of the test suite.
+ """
+ assert_deepequal(
+ self.entity_attrs,
+ api.Command['%s_show' % self.entity_class](self.entity_pkey)
+ )
+
+
+class TestCertManipCmdUser(CertManipCmdTestBase):
+ entity_class = 'user'
+ entity_pkey = u'tuser'
+ entity_subject = entity_pkey
+ entity_principal = u'tuser'
+ non_existent_entity = u'nonexistentuser'
+
+ cmd_options = dict(
+ entity_add=dict(givenname=u'Test', sn=u'User'),
+ caacl=dict(user=[u'tuser']),
+ )
+
+ cert_add_cmd = api.Command.user_add_cert
+ cert_del_cmd = api.Command.user_remove_cert
+
+ cert_add_summary = u'Added certificates to user "%s"'
+ cert_del_summary = u'Removed certificates from user "%s"'
+
+ @classmethod
+ def add_caacl(cls):
+ api.Command['caacl_add_%s' % cls.entity_class](
+ cls.default_caacl, **cls.cmd_options['caacl'])
+
+ @classmethod
+ def remove_caacl(cls):
+ api.Command['caacl_remove_%s' % cls.entity_class](
+ cls.default_caacl, **cls.cmd_options['caacl'])
+
+
+class TestCertManipCmdHost(CertManipCmdTestBase):
+ entity_class = 'host'
+ entity_pkey = u'host.example.com'
+ entity_subject = entity_pkey
+ entity_principal = u'host/%s' % entity_pkey
+ non_existent_entity = u'non.existent.host.com'
+
+ cmd_options = dict(
+ entity_add=dict(force=True),
+ )
+
+ cert_add_cmd = api.Command.host_add_cert
+ cert_del_cmd = api.Command.host_remove_cert
+
+ cert_add_summary = u'Added certificates to host "%s"'
+ cert_del_summary = u'Removed certificates from host "%s"'
+
+
+class TestCertManipCmdService(CertManipCmdTestBase):
+ entity_class = 'service'
+ entity_pkey = u'testservice/%s@%s' % (TestCertManipCmdHost.entity_pkey,
+ api.env.realm)
+ entity_subject = TestCertManipCmdHost.entity_pkey
+ entity_principal = entity_pkey
+ non_existent_entity = u'testservice/non.existent.host.com'
+
+ cmd_options = dict(
+ entity_add=dict(force=True),
+ )
+
+ cert_add_cmd = api.Command.service_add_cert
+ cert_del_cmd = api.Command.service_remove_cert
+
+ cert_add_summary = u'Added certificates to service principal "%s"'
+ cert_del_summary = u'Removed certificates from service principal "%s"'
+
+ @classmethod
+ def add_entity(cls):
+ api.Command.host_add(TestCertManipCmdHost.entity_pkey, force=True)
+ super(TestCertManipCmdService, cls).add_entity()
+
+ @classmethod
+ def delete_entity(cls):
+ super(TestCertManipCmdService, cls).delete_entity()
+ try:
+ api.Command.host_del(TestCertManipCmdHost.entity_pkey)
+ except errors.NotFound:
+ pass
--
2.4.3
From 9cb2d4cea54167cff8f512bc140c7d3f58c7901c Mon Sep 17 00:00:00 2001
From: Martin Babinsky <mbabi...@redhat.com>
Date: Tue, 23 Jun 2015 13:42:01 +0200
Subject: [PATCH 3/4] new commands to manage user/host/service certificates
A new group of commands is introduced that simplifies adding and removing
binary certificates to entries. A general form of the command is
ipa [user/host/service]-[add/remove]-cert [pkey] --certificate=[BASE64 BLOB]
Part of http://www.freeipa.org/page/V4/User_Certificates and
https://fedorahosted.org/freeipa/ticket/4238
---
API.txt | 66 +++++++++++++++++++++++++++++++++++++++++++++++
VERSION | 4 +--
ipalib/plugins/host.py | 27 +++++++++++++++++--
ipalib/plugins/service.py | 21 +++++++++++++++
ipalib/plugins/user.py | 52 +++++++++++++++++++++++++++++++++++++
5 files changed, 166 insertions(+), 4 deletions(-)
diff --git a/API.txt b/API.txt
index bccebe55da8a785cbb6ca782904d7523c4a9322f..3ba67be60549eed860ff9e8d600cf02dc50fa456 100644
--- a/API.txt
+++ b/API.txt
@@ -2066,6 +2066,17 @@ option: Str('version?', exclude='webui')
output: Entry('result', <type 'dict'>, Gettext('A dictionary representing an LDAP entry', domain='ipa', localedir=None))
output: Output('summary', (<type 'unicode'>, <type 'NoneType'>), None)
output: PrimaryKey('value', None, None)
+command: host_add_cert
+args: 1,5,3
+arg: Str('fqdn', attribute=True, cli_name='hostname', multivalue=False, primary_key=True, query=True, required=True)
+option: Flag('all', autofill=True, cli_name='all', default=False, exclude='webui')
+option: Flag('no_members', autofill=True, default=False, exclude='webui')
+option: Flag('raw', autofill=True, cli_name='raw', default=False, exclude='webui')
+option: Bytes('usercertificate', alwaysask=True, attribute=True, cli_name='certificate', multivalue=True, required=False)
+option: Str('version?', exclude='webui')
+output: Entry('result', <type 'dict'>, Gettext('A dictionary representing an LDAP entry', domain='ipa', localedir=None))
+output: Output('summary', (<type 'unicode'>, <type 'NoneType'>), None)
+output: PrimaryKey('value', None, None)
command: host_add_managedby
args: 1,5,3
arg: Str('fqdn', attribute=True, cli_name='hostname', multivalue=False, primary_key=True, query=True, required=True)
@@ -2220,6 +2231,17 @@ option: Str('version?', exclude='webui')
output: Entry('result', <type 'dict'>, Gettext('A dictionary representing an LDAP entry', domain='ipa', localedir=None))
output: Output('summary', (<type 'unicode'>, <type 'NoneType'>), None)
output: PrimaryKey('value', None, None)
+command: host_remove_cert
+args: 1,5,3
+arg: Str('fqdn', attribute=True, cli_name='hostname', multivalue=False, primary_key=True, query=True, required=True)
+option: Flag('all', autofill=True, cli_name='all', default=False, exclude='webui')
+option: Flag('no_members', autofill=True, default=False, exclude='webui')
+option: Flag('raw', autofill=True, cli_name='raw', default=False, exclude='webui')
+option: Bytes('usercertificate', alwaysask=True, attribute=True, cli_name='certificate', multivalue=True, required=False)
+option: Str('version?', exclude='webui')
+output: Entry('result', <type 'dict'>, Gettext('A dictionary representing an LDAP entry', domain='ipa', localedir=None))
+output: Output('summary', (<type 'unicode'>, <type 'NoneType'>), None)
+output: PrimaryKey('value', None, None)
command: host_remove_managedby
args: 1,5,3
arg: Str('fqdn', attribute=True, cli_name='hostname', multivalue=False, primary_key=True, query=True, required=True)
@@ -3851,6 +3873,17 @@ option: Str('version?', exclude='webui')
output: Entry('result', <type 'dict'>, Gettext('A dictionary representing an LDAP entry', domain='ipa', localedir=None))
output: Output('summary', (<type 'unicode'>, <type 'NoneType'>), None)
output: PrimaryKey('value', None, None)
+command: service_add_cert
+args: 1,5,3
+arg: Str('krbprincipalname', attribute=True, cli_name='principal', multivalue=False, primary_key=True, query=True, required=True)
+option: Flag('all', autofill=True, cli_name='all', default=False, exclude='webui')
+option: Flag('no_members', autofill=True, default=False, exclude='webui')
+option: Flag('raw', autofill=True, cli_name='raw', default=False, exclude='webui')
+option: Bytes('usercertificate', alwaysask=True, attribute=True, cli_name='certificate', multivalue=True, required=False)
+option: Str('version?', exclude='webui')
+output: Entry('result', <type 'dict'>, Gettext('A dictionary representing an LDAP entry', domain='ipa', localedir=None))
+output: Output('summary', (<type 'unicode'>, <type 'NoneType'>), None)
+output: PrimaryKey('value', None, None)
command: service_add_host
args: 1,5,3
arg: Str('krbprincipalname', attribute=True, cli_name='principal', multivalue=False, primary_key=True, query=True, required=True)
@@ -3969,6 +4002,17 @@ option: Str('version?', exclude='webui')
output: Entry('result', <type 'dict'>, Gettext('A dictionary representing an LDAP entry', domain='ipa', localedir=None))
output: Output('summary', (<type 'unicode'>, <type 'NoneType'>), None)
output: PrimaryKey('value', None, None)
+command: service_remove_cert
+args: 1,5,3
+arg: Str('krbprincipalname', attribute=True, cli_name='principal', multivalue=False, primary_key=True, query=True, required=True)
+option: Flag('all', autofill=True, cli_name='all', default=False, exclude='webui')
+option: Flag('no_members', autofill=True, default=False, exclude='webui')
+option: Flag('raw', autofill=True, cli_name='raw', default=False, exclude='webui')
+option: Bytes('usercertificate', alwaysask=True, attribute=True, cli_name='certificate', multivalue=True, required=False)
+option: Str('version?', exclude='webui')
+output: Entry('result', <type 'dict'>, Gettext('A dictionary representing an LDAP entry', domain='ipa', localedir=None))
+output: Output('summary', (<type 'unicode'>, <type 'NoneType'>), None)
+output: PrimaryKey('value', None, None)
command: service_remove_host
args: 1,5,3
arg: Str('krbprincipalname', attribute=True, cli_name='principal', multivalue=False, primary_key=True, query=True, required=True)
@@ -5156,6 +5200,17 @@ option: Str('version?', exclude='webui')
output: Entry('result', <type 'dict'>, Gettext('A dictionary representing an LDAP entry', domain='ipa', localedir=None))
output: Output('summary', (<type 'unicode'>, <type 'NoneType'>), None)
output: PrimaryKey('value', None, None)
+command: user_add_cert
+args: 1,5,3
+arg: Str('uid', attribute=True, cli_name='login', maxlength=255, multivalue=False, pattern='^[a-zA-Z0-9_.][a-zA-Z0-9_.-]{0,252}[a-zA-Z0-9_.$-]?$', primary_key=True, query=True, required=True)
+option: Flag('all', autofill=True, cli_name='all', default=False, exclude='webui')
+option: Flag('no_members', autofill=True, default=False, exclude='webui')
+option: Flag('raw', autofill=True, cli_name='raw', default=False, exclude='webui')
+option: Bytes('usercertificate', alwaysask=True, attribute=True, cli_name='certificate', multivalue=True, required=False)
+option: Str('version?', exclude='webui')
+output: Entry('result', <type 'dict'>, Gettext('A dictionary representing an LDAP entry', domain='ipa', localedir=None))
+output: Output('summary', (<type 'unicode'>, <type 'NoneType'>), None)
+output: PrimaryKey('value', None, None)
command: user_del
args: 1,4,3
arg: Str('uid', attribute=True, cli_name='login', maxlength=255, multivalue=True, pattern='^[a-zA-Z0-9_.][a-zA-Z0-9_.-]{0,252}[a-zA-Z0-9_.$-]?$', primary_key=True, query=True, required=True)
@@ -5295,6 +5350,17 @@ option: Str('version?', exclude='webui')
output: Entry('result', <type 'dict'>, Gettext('A dictionary representing an LDAP entry', domain='ipa', localedir=None))
output: Output('summary', (<type 'unicode'>, <type 'NoneType'>), None)
output: PrimaryKey('value', None, None)
+command: user_remove_cert
+args: 1,5,3
+arg: Str('uid', attribute=True, cli_name='login', maxlength=255, multivalue=False, pattern='^[a-zA-Z0-9_.][a-zA-Z0-9_.-]{0,252}[a-zA-Z0-9_.$-]?$', primary_key=True, query=True, required=True)
+option: Flag('all', autofill=True, cli_name='all', default=False, exclude='webui')
+option: Flag('no_members', autofill=True, default=False, exclude='webui')
+option: Flag('raw', autofill=True, cli_name='raw', default=False, exclude='webui')
+option: Bytes('usercertificate', alwaysask=True, attribute=True, cli_name='certificate', multivalue=True, required=False)
+option: Str('version?', exclude='webui')
+output: Entry('result', <type 'dict'>, Gettext('A dictionary representing an LDAP entry', domain='ipa', localedir=None))
+output: Output('summary', (<type 'unicode'>, <type 'NoneType'>), None)
+output: PrimaryKey('value', None, None)
command: user_show
args: 1,5,3
arg: Str('uid', attribute=True, cli_name='login', maxlength=255, multivalue=False, pattern='^[a-zA-Z0-9_.][a-zA-Z0-9_.-]{0,252}[a-zA-Z0-9_.$-]?$', primary_key=True, query=True, required=True)
diff --git a/VERSION b/VERSION
index 2f884ff73afad57f35f06ce279add5c078073353..266a04af1a61132637112611b7e86649ff818c2a 100644
--- a/VERSION
+++ b/VERSION
@@ -90,5 +90,5 @@ IPA_DATA_VERSION=20100614120000
# #
########################################################
IPA_API_VERSION_MAJOR=2
-IPA_API_VERSION_MINOR=136
-# Last change: pvoborni: add topologysuffix-verify command
+IPA_API_VERSION_MINOR=137
+# Last change: mbabinsk: Commands to manage user/host/service certificates
diff --git a/ipalib/plugins/host.py b/ipalib/plugins/host.py
index f5871f81e49c0bf49fd2cba14fdde7d1b570e462..410b4bd120743a6ad5787fbd2a55534b4f108601 100644
--- a/ipalib/plugins/host.py
+++ b/ipalib/plugins/host.py
@@ -28,11 +28,12 @@ from ipalib.plugins.baseldap import (LDAPQuery, LDAPObject, LDAPCreate,
LDAPDelete, LDAPUpdate, LDAPSearch,
LDAPRetrieve, LDAPAddMember,
LDAPRemoveMember, host_is_master,
- pkey_to_value, add_missing_object_class)
+ pkey_to_value, add_missing_object_class,
+ LDAPAddAttribute, LDAPRemoveAttribute)
from ipalib.plugins.service import (split_principal, validate_certificate,
set_certificate_attrs, ticket_flags_params, update_krbticketflags,
set_kerberos_attrs, rename_ipaallowedtoperform_from_ldap,
- rename_ipaallowedtoperform_to_ldap)
+ rename_ipaallowedtoperform_to_ldap, revoke_certs)
from ipalib.plugins.dns import (dns_container_exists, _record_types,
add_records_for_host_validation, add_records_for_host,
get_reverse_zone)
@@ -1246,3 +1247,25 @@ class host_disallow_create_keytab(LDAPRemoveMember):
rename_ipaallowedtoperform_from_ldap(entry_attrs, options)
rename_ipaallowedtoperform_from_ldap(failed, options)
return (completed, dn)
+
+
+@register()
+class host_add_cert(LDAPAddAttribute):
+ __doc__ = _('Add certificates to host entry')
+ msg_summary = _('Added certificates to host "%(value)s"')
+ attribute = 'usercertificate'
+
+
+@register()
+class host_remove_cert(LDAPRemoveAttribute):
+ __doc__ = _('Remove certificates from host entry')
+ msg_summary = _('Removed certificates from host "%(value)s"')
+ attribute = 'usercertificate'
+
+ def post_callback(self, ldap, dn, entry_attrs, *keys, **options):
+ assert isinstance(dn, DN)
+
+ if 'usercertificate' in options:
+ revoke_certs(options['usercertificate'], self.log)
+
+ return dn
diff --git a/ipalib/plugins/service.py b/ipalib/plugins/service.py
index 18d7b3e549246a45ae72a2516ef69085c4c66c83..a5e10921beac8b232f6b74640ef17713f7297a3d 100644
--- a/ipalib/plugins/service.py
+++ b/ipalib/plugins/service.py
@@ -861,3 +861,24 @@ class service_disable(LDAPQuery):
value=pkey_to_value(keys[0], options),
)
+
+@register()
+class service_add_cert(LDAPAddAttribute):
+ __doc__ = _('Add new certificates to a service')
+ msg_summary = _('Added certificates to service principal "%(value)s"')
+ attribute = 'usercertificate'
+
+
+@register()
+class service_remove_cert(LDAPRemoveAttribute):
+ __doc__ = _('Remove certificates from a service')
+ msg_summary = _('Removed certificates from service principal "%(value)s"')
+ attribute = 'usercertificate'
+
+ def post_callback(self, ldap, dn, entry_attrs, *keys, **options):
+ assert isinstance(dn, DN)
+
+ if 'usercertificate' in options:
+ revoke_certs(options['usercertificate'], self.log)
+
+ return dn
diff --git a/ipalib/plugins/user.py b/ipalib/plugins/user.py
index bc6989ccee40b10fb0a78cccdae3fc6d41bdfa29..9bd7bf7e5242234ead4c39a6346e57865b2e2124 100644
--- a/ipalib/plugins/user.py
+++ b/ipalib/plugins/user.py
@@ -1001,3 +1001,55 @@ class user_status(LDAPQuery):
summary=unicode(_('Account disabled: %(disabled)s' %
dict(disabled=disabled))),
)
+
+
+@register()
+class user_add_cert(LDAPAddAttribute):
+ __doc__ = _('Add one or more certificates to the user entry')
+ msg_summary = _('Added certificates to user "%(value)s"')
+ attribute = 'usercertificate'
+
+ def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys,
+ **options):
+ assert isinstance(dn, DN)
+
+ new_attr_name = '%s;binary' % self.attribute
+ if self.attribute in entry_attrs:
+ entry_attrs[new_attr_name] = entry_attrs.pop(self.attribute)
+
+ return dn
+
+ def post_callback(self, ldap, dn, entry_attrs, *keys, **options):
+ assert isinstance(dn, DN)
+
+ old_attr_name = '%s;binary' % self.attribute
+ if old_attr_name in entry_attrs:
+ entry_attrs[self.attribute] = entry_attrs.pop(old_attr_name)
+
+ return dn
+
+
+@register()
+class user_remove_cert(LDAPRemoveAttribute):
+ __doc__ = _('Remove one or more certificates to the user entry')
+ msg_summary = _('Removed certificates from user "%(value)s"')
+ attribute = 'usercertificate'
+
+ def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys,
+ **options):
+ assert isinstance(dn, DN)
+
+ new_attr_name = '%s;binary' % self.attribute
+ if self.attribute in entry_attrs:
+ entry_attrs[new_attr_name] = entry_attrs.pop(self.attribute)
+
+ return dn
+
+ def post_callback(self, ldap, dn, entry_attrs, *keys, **options):
+ assert isinstance(dn, DN)
+
+ old_attr_name = '%s;binary' % self.attribute
+ if old_attr_name in entry_attrs:
+ entry_attrs[self.attribute] = entry_attrs.pop(old_attr_name)
+
+ return dn
--
2.4.3
From 9583c2a13732c841f44767626625b7ffa2a006b0 Mon Sep 17 00:00:00 2001
From: Martin Babinsky <mbabi...@redhat.com>
Date: Tue, 23 Jun 2015 13:40:30 +0200
Subject: [PATCH 2/4] reworked certificate normalization and revocation
Validation of certificate is now handled by `x509.validate_certificate'.
Revocation of the host and service certificates was factored out to a separate
function.
Part of http://www.freeipa.org/page/V4/User_Certificates
---
ipalib/plugins/host.py | 75 +++----------------------------
ipalib/plugins/service.py | 112 ++++++++++++++++------------------------------
ipalib/x509.py | 14 ++++--
3 files changed, 55 insertions(+), 146 deletions(-)
diff --git a/ipalib/plugins/host.py b/ipalib/plugins/host.py
index e81dca94e124b080a3d68a3b1cfd079710e30336..f5871f81e49c0bf49fd2cba14fdde7d1b570e462 100644
--- a/ipalib/plugins/host.py
+++ b/ipalib/plugins/host.py
@@ -786,30 +786,8 @@ class host_del(LDAPDelete):
entry_attrs = ldap.get_entry(dn, ['usercertificate'])
except errors.NotFound:
self.obj.handle_not_found(*keys)
- for cert in entry_attrs.get('usercertificate', []):
- cert = x509.normalize_certificate(cert)
- try:
- serial = unicode(x509.get_serial_number(cert, x509.DER))
- try:
- result = api.Command['cert_show'](serial)['result']
- if 'revocation_reason' not in result:
- try:
- api.Command['cert_revoke'](serial,
- revocation_reason=4)
- except errors.NotImplementedError:
- # some CA's might not implement revoke
- pass
- except errors.NotImplementedError:
- # some CA's might not implement revoke
- pass
- except NSPRError, nsprerr:
- if nsprerr.errno == -8183:
- # If we can't decode the cert them proceed with
- # removing the host.
- self.log.info("Problem decoding certificate %s" %
- nsprerr.args[1])
- else:
- raise nsprerr
+
+ revoke_certs(entry_attrs.get('usercertificate', []), self.log)
return dn
@@ -879,29 +857,8 @@ class host_mod(LDAPUpdate):
old_certs = entry_attrs_old.get('usercertificate', [])
old_certs_der = map(x509.normalize_certificate, old_certs)
removed_certs_der = set(old_certs_der) - set(certs_der)
- for cert in removed_certs_der:
- try:
- serial = unicode(x509.get_serial_number(cert, x509.DER))
- try:
- result = api.Command['cert_show'](serial)['result']
- if 'revocation_reason' not in result:
- try:
- api.Command['cert_revoke'](
- serial, revocation_reason=4)
- except errors.NotImplementedError:
- # some CA's might not implement revoke
- pass
- except errors.NotImplementedError:
- # some CA's might not implement revoke
- pass
- except NSPRError, nsprerr:
- if nsprerr.errno == -8183:
- # If we can't decode the cert them proceed with
- # modifying the host.
- self.log.info("Problem decoding certificate %s" %
- nsprerr.args[1])
- else:
- raise nsprerr
+ revoke_certs(removed_certs_der, self.log)
+
if certs:
entry_attrs['usercertificate'] = certs_der
@@ -1163,31 +1120,9 @@ class host_disable(LDAPQuery):
self.obj.handle_not_found(*keys)
if self.api.Command.ca_is_enabled()['result']:
certs = entry_attrs.get('usercertificate', [])
- for cert in map(x509.normalize_certificate, certs):
- try:
- serial = unicode(x509.get_serial_number(cert, x509.DER))
- try:
- result = api.Command['cert_show'](serial)['result']
- if 'revocation_reason' not in result:
- try:
- api.Command['cert_revoke'](serial,
- revocation_reason=4)
- except errors.NotImplementedError:
- # some CA's might not implement revoke
- pass
- except errors.NotImplementedError:
- # some CA's might not implement revoke
- pass
- except NSPRError, nsprerr:
- if nsprerr.errno == -8183:
- # If we can't decode the cert them proceed with
- # disabling the host.
- self.log.info("Problem decoding certificate %s" %
- nsprerr.args[1])
- else:
- raise nsprerr
if certs:
+ revoke_certs(certs, self.log)
# Remove the usercertificate altogether
entry_attrs['usercertificate'] = None
ldap.update_entry(entry_attrs)
diff --git a/ipalib/plugins/service.py b/ipalib/plugins/service.py
index 166d978a248e7c5da6f8df4b534edad0a0799b7e..18d7b3e549246a45ae72a2516ef69085c4c66c83 100644
--- a/ipalib/plugins/service.py
+++ b/ipalib/plugins/service.py
@@ -238,16 +238,42 @@ def normalize_principal(principal):
def validate_certificate(ugettext, cert):
"""
- For now just verify that it is properly base64-encoded.
+ Check whether the certificate is properly encoded to DER
"""
- if cert and util.isvalid_base64(cert):
+ if api.env.in_server:
+ x509.validate_certificate(cert, datatype=x509.DER)
+
+
+def revoke_certs(certs, logger=None):
+ """
+ revoke the certificates removed from host/service entry
+ """
+ for cert in certs:
+ try:
+ cert = x509.normalize_certificate(cert)
+ except errors.CertificateFormatError as e:
+ if logger is not None:
+ logger.info("Problem decoding certificate: %s" % e)
+
+ serial = unicode(x509.get_serial_number(cert, x509.DER))
+
+ try:
+ result = api.Command['cert_show'](unicode(serial))['result']
+ except errors.CertificateOperationError:
+ continue
+ if 'revocation_reason' in result:
+ continue
+ if x509.normalize_certificate(result['certificate']) != cert:
+ continue
+
try:
- base64.b64decode(cert)
- except Exception, e:
- raise errors.Base64DecodeError(reason=str(e))
- else:
- # We'll assume this is DER data
- pass
+ api.Command['cert_revoke'](unicode(serial),
+ revocation_reason=4)
+ except errors.NotImplementedError:
+ # some CA's might not implement revoke
+ pass
+
+
def set_certificate_attrs(entry_attrs):
"""
@@ -566,27 +592,8 @@ class service_del(LDAPDelete):
entry_attrs = ldap.get_entry(dn, ['usercertificate'])
except errors.NotFound:
self.obj.handle_not_found(*keys)
- for cert in entry_attrs.get('usercertificate', []):
- try:
- serial = unicode(x509.get_serial_number(cert, x509.DER))
- try:
- result = api.Command['cert_show'](unicode(serial))['result']
- if 'revocation_reason' not in result:
- try:
- api.Command['cert_revoke'](unicode(serial), revocation_reason=4)
- except errors.NotImplementedError:
- # some CA's might not implement revoke
- pass
- except errors.NotImplementedError:
- # some CA's might not implement revoke
- pass
- except NSPRError, nsprerr:
- if nsprerr.errno == -8183:
- # If we can't decode the cert them proceed with
- # removing the service.
- self.log.info("Problem decoding certificate %s" % nsprerr.args[1])
- else:
- raise nsprerr
+ revoke_certs(entry_attrs.get('usercertificate', []), self.log)
+
return dn
@@ -622,29 +629,8 @@ class service_mod(LDAPUpdate):
old_certs = entry_attrs_old.get('usercertificate', [])
old_certs_der = map(x509.normalize_certificate, old_certs)
removed_certs_der = set(old_certs_der) - set(certs_der)
- for cert in removed_certs_der:
- try:
- serial = unicode(x509.get_serial_number(cert, x509.DER))
- try:
- result = api.Command['cert_show'](serial)['result']
- if 'revocation_reason' not in result:
- try:
- api.Command['cert_revoke'](
- serial, revocation_reason=4)
- except errors.NotImplementedError:
- # some CA's might not implement revoke
- pass
- except errors.NotImplementedError:
- # some CA's might not implement revoke
- pass
- except NSPRError, nsprerr:
- if nsprerr.errno == -8183:
- # If we can't decode the cert them proceed with
- # modifying the host.
- self.log.info("Problem decoding certificate %s" %
- nsprerr.args[1])
- else:
- raise nsprerr
+ revoke_certs(removed_certs_der, self.log)
+
if certs:
entry_attrs['usercertificate'] = certs_der
@@ -854,29 +840,9 @@ class service_disable(LDAPQuery):
if self.api.Command.ca_is_enabled()['result']:
certs = entry_attrs.get('usercertificate', [])
- for cert in map(x509.normalize_certificate, certs):
- try:
- serial = unicode(x509.get_serial_number(cert, x509.DER))
- try:
- result = api.Command['cert_show'](unicode(serial))['result']
- if 'revocation_reason' not in result:
- try:
- api.Command['cert_revoke'](unicode(serial), revocation_reason=4)
- except errors.NotImplementedError:
- # some CA's might not implement revoke
- pass
- except errors.NotImplementedError:
- # some CA's might not implement revoke
- pass
- except NSPRError, nsprerr:
- if nsprerr.errno == -8183:
- # If we can't decode the cert them proceed with
- # disabling the service
- self.log.info("Problem decoding certificate %s" % nsprerr.args[1])
- else:
- raise nsprerr
if len(certs) > 0:
+ revoke_certs(certs, self.log)
# Remove the usercertificate altogether
entry_attrs['usercertificate'] = None
ldap.update_entry(entry_attrs)
diff --git a/ipalib/x509.py b/ipalib/x509.py
index a87dbf4130c60b1b1daf8bbb2ffb81c208f2529c..edd73ebdc3b3732d326cd8f414bc957f1e4deb87 100644
--- a/ipalib/x509.py
+++ b/ipalib/x509.py
@@ -294,16 +294,24 @@ def normalize_certificate(rawcert):
# was base64-encoded and now its not or it came in as DER format.
# Let's decode it and see. Fetching the serial number will pass the
# certificate through the NSS DER parser.
+ validate_certificate(dercert, datatype=DER)
+
+ return dercert
+
+
+def validate_certificate(cert, datatype=PEM, dbdir=None):
+ """
+ Perform certificate validation by trying to load it into NSS database
+ """
try:
- serial = unicode(get_serial_number(dercert, DER))
- except NSPRError, nsprerr:
+ load_certificate(cert, datatype=datatype, dbdir=dbdir)
+ except NSPRError as nsprerr:
if nsprerr.errno == -8183: # SEC_ERROR_BAD_DER
raise errors.CertificateFormatError(
error=_('improperly formatted DER-encoded certificate'))
else:
raise errors.CertificateFormatError(error=str(nsprerr))
- return dercert
def write_certificate(rawcert, filename):
"""
--
2.4.3
From 93e0ef7079db98af9d3f5d184fb202755d06e75c Mon Sep 17 00:00:00 2001
From: Martin Babinsky <mbabi...@redhat.com>
Date: Tue, 23 Jun 2015 13:39:35 +0200
Subject: [PATCH 1/4] baseldap: add support for API commands managing only a
single attribute
This patch extends the API framework with a set of classes which add/remove
values to a single LDAPObject attribute.
---
ipalib/plugins/baseldap.py | 114 +++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 114 insertions(+)
diff --git a/ipalib/plugins/baseldap.py b/ipalib/plugins/baseldap.py
index 2eab69f3decb3359e82d30a0a3a595e81a6d9bc3..2309bbc00dfba362263e37edddb7b374ef40447d 100644
--- a/ipalib/plugins/baseldap.py
+++ b/ipalib/plugins/baseldap.py
@@ -2367,3 +2367,117 @@ class LDAPRemoveReverseMember(LDAPModReverseMember):
def interactive_prompt_callback(self, kw):
return
+
+
+class LDAPModAttribute(LDAPQuery):
+
+ attribute = None
+
+ has_output = output.standard_entry
+
+ def get_options(self):
+ for option in super(LDAPModAttribute, self).get_options():
+ yield option
+
+ option = self.obj.params[self.attribute]
+ attribute = 'virtual_attribute' not in option.flags
+ yield option.clone(attribute=attribute, alwaysask=True)
+
+ def _update_attrs(self, update, entry_attrs):
+ raise NotImplementedError("%s.update_attrs()", self.__class__.__name__)
+
+ def execute(self, *keys, **options):
+ ldap = self.obj.backend
+
+ dn = self.obj.get_dn(*keys, **options)
+ entry_attrs = ldap.make_entry(dn, self.args_options_2_entry(**options))
+
+ if options.get('all', False):
+ attrs_list = ['*', self.obj.primary_key.name]
+ else:
+ attrs_list = {self.obj.primary_key.name}
+ attrs_list.update(entry_attrs.keys())
+ attrs_list = list(attrs_list)
+
+ for callback in self.get_callbacks('pre'):
+ entry_attrs.dn = callback(
+ self, ldap, entry_attrs.dn, entry_attrs, attrs_list,
+ *keys, **options)
+
+ try:
+ update = self._exc_wrapper(keys, options, ldap.get_entry)(
+ entry_attrs.dn, entry_attrs.keys())
+ self._update_attrs(update, entry_attrs)
+
+ self._exc_wrapper(keys, options, ldap.update_entry)(update)
+ except errors.NotFound:
+ self.obj.handle_not_found(*keys)
+
+ try:
+ entry_attrs = self._exc_wrapper(keys, options, ldap.get_entry)(
+ entry_attrs.dn, attrs_list)
+ except errors.NotFound:
+ raise errors.MidairCollision(
+ format=_('the entry was deleted while being modified')
+ )
+
+ for callback in self.get_callbacks('post'):
+ entry_attrs.dn = callback(
+ self, ldap, entry_attrs.dn, entry_attrs, *keys, **options)
+
+ entry_attrs = entry_to_dict(entry_attrs, **options)
+
+ if self.obj.primary_key:
+ pkey = keys[-1]
+ else:
+ pkey = None
+
+ return dict(result=entry_attrs, value=pkey_to_value(pkey, options))
+
+ def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys,
+ **options):
+ assert isinstance(dn, DN)
+ return dn
+
+ def post_callback(self, ldap, dn, entry_attrs, *keys, **options):
+ assert isinstance(dn, DN)
+ return dn
+
+ def exc_callback(self, keys, options, exc, call_func, *call_args,
+ **call_kwargs):
+ raise exc
+
+ def interactive_prompt_callback(self, kw):
+ return
+
+
+class LDAPAddAttribute(LDAPModAttribute):
+ msg_summary = _('added attribute value to entry %(value)')
+
+ def _update_attrs(self, update, entry_attrs):
+ for name, value in entry_attrs.iteritems():
+ old_value = set(update.get(name, []))
+ value_to_add = set(value)
+
+ if not old_value.isdisjoint(value_to_add):
+ raise errors.ExecutionError(
+ message=_('\'%s\' already contains one or more values'
+ % name)
+ )
+
+ update[name] = list(old_value | value_to_add)
+
+
+class LDAPRemoveAttribute(LDAPModAttribute):
+ msg_summary = _('removed attribute values from entry %(value)')
+
+ def _update_attrs(self, update, entry_attrs):
+ for name, value in entry_attrs.iteritems():
+ old_value = set(update.get(name, []))
+ value_to_remove = set(value)
+
+ if not value_to_remove.issubset(old_value):
+ raise errors.AttrValueNotFound(
+ attr=name, value=_("one or more values to remove"))
+
+ update[name] = list(old_value - value_to_remove)
--
2.4.3
--
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
