generalize certificate creation during testing

With added support for multiple certificates for hosts, services, and even
users, IPA testing framework will need a more flexible way to generate
temporary testing certificates for these entities. This patch modifies the
currently used `testcert` module to support these requirements.

Related to work on http://www.freeipa.org/page/V4/User_Certificates

Reviewed-By: Milan Kubík <mkubik@redhat.com>

3 files changed, 26 insertions, 26 deletions

diff --git a/ipatests/test_xmlrpc/test_host_plugin.py b/ipatests/test_xmlrpc/test_host_plugin.py
index e46e502d2..4c574b957 100644
--- a/ipatests/test_xmlrpc/test_host_plugin.py
+++ b/ipatests/test_xmlrpc/test_host_plugin.py
@@ -94,6 +94,9 @@ hostgroup1 = u'testhostgroup1'
 hostgroup1_dn = DN(('cn',hostgroup1),('cn','hostgroups'),('cn','accounts'),
                     api.env.basedn)
 
+host_cert = get_testcert(DN(('CN', api.env.host), x509.subject_base()),
+                         'host/%s@%s' % (api.env.host, api.env.realm))
+
 
 class HostTracker(Tracker):
     """Wraps and tracks modifications to a Host object
@@ -360,10 +363,10 @@ class TestCRUD(XMLRPC_test):
     def test_update_simple(self, host):
         host.update(dict(
                         description=u'Updated host 1',
-                        usercertificate=get_testcert()),
+                        usercertificate=host_cert),
                     expected_updates=dict(
                         description=[u'Updated host 1'],
-                        usercertificate=[base64.b64decode(get_testcert())],
+                        usercertificate=[base64.b64decode(host_cert)],
                         issuer=fuzzy_issuer,
                         md5_fingerprint=fuzzy_hash,
                         serial_number=fuzzy_digits,
diff --git a/ipatests/test_xmlrpc/test_service_plugin.py b/ipatests/test_xmlrpc/test_service_plugin.py
index 946dc572b..006c35999 100644
--- a/ipatests/test_xmlrpc/test_service_plugin.py
+++ b/ipatests/test_xmlrpc/test_service_plugin.py
@@ -46,6 +46,8 @@ host3dn = DN(('fqdn',fqdn3),('cn','computers'),('cn','accounts'),api.env.basedn)
 role1 = u'Test Role'
 role1_dn = DN(('cn', role1), api.env.container_rolegroup, api.env.basedn)
 
+servercert= get_testcert(DN(('CN', api.env.host), x509.subject_base()),
+                         'unittest/%s@%s' % (api.env.host, api.env.realm))
 badservercert = 'MIICbzCCAdigAwIBAgICA/4wDQYJKoZIhvcNAQEFBQAwKTEnMCUGA1UEAxMeSVBBIFRlc3QgQ2VydGlmaWNhdGUgQXV0aG9yaXR5MB4XDTEwMDgwOTE1MDIyN1oXDTIwMDgwOTE1MDIyN1owKTEMMAoGA1UEChMDSVBBMRkwFwYDVQQDExBwdW1hLmdyZXlvYWsuY29tMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAwYbfEOQPgGenPn9vt1JFKvWm/Je3y2tawGWA3LXDuqfFJyYtZ8ib3TcBUOnLk9WK5g2qCwHaNlei7bj8ggIfr5hegAVe10cun+wYErjnYo7hsHYd+57VZezeipWrXu+7NoNd4+c4A5lk4A/xJay9j3bYx2oOM8BEox4xWYoWge1ljPrc5JK46f0X7AGW4F2VhnKPnf8rwSuzI1U8VGjutyM9TWNy3m9KMWeScjyG/ggIpOjUDMV7HkJL0Di61lznR9jXubpiEC7gWGbTp84eGl/Nn9bgK1AwHfJ2lHwfoY4uiL7ge1gyP6EvuUlHoBzdb7pekiX28iePjW3iEG9IawIDAQABoyIwIDARBglghkgBhvhCAQEEBAMCBkAwCwYDVR0PBAQDAgUgMA0GCSqGSIb3DQEBBQUAA4GBACRESLemRV9BPxfEgbALuxH5oE8jQm8WZ3pm2pALbpDlAd9wQc3yVf6RtkfVthyDnM18bg7IhxKpd77/p3H8eCnS8w5MLVRda6ktUC6tGhFTS4QKAf0WyDGTcIgkXbeDw0OPAoNHivoXbIXIIRxlw/XgaSaMzJQDBG8iROsN4kCv'
 
 user1 = u'tuser1'
@@ -78,7 +80,7 @@ class test_service(Declarative):
 
         dict(
             desc='Try to update non-existent %r' % service1,
-            command=('service_mod', [service1], dict(usercertificate=get_testcert())),
+            command=('service_mod', [service1], dict(usercertificate=servercert)),
             expected=errors.NotFound(
                 reason=u'%s: service not found' % service1),
         ),
@@ -390,12 +392,12 @@ class test_service(Declarative):
 
         dict(
             desc='Update %r' % service1,
-            command=('service_mod', [service1], dict(usercertificate=get_testcert())),
+            command=('service_mod', [service1], dict(usercertificate=servercert)),
             expected=dict(
                 value=service1,
                 summary=u'Modified service "%s"' % service1,
                 result=dict(
-                    usercertificate=[base64.b64decode(get_testcert())],
+                    usercertificate=[base64.b64decode(servercert)],
                     krbprincipalname=[service1],
                     managedby_host=[fqdn1],
                     valid_not_before=fuzzy_date,
@@ -430,7 +432,7 @@ class test_service(Declarative):
                 value=service1,
                 summary=u'Modified service "%s"' % service1,
                 result=dict(
-                    usercertificate=[base64.b64decode(get_testcert())],
+                    usercertificate=[base64.b64decode(servercert)],
                     krbprincipalname=[service1],
                     managedby_host=[fqdn1],
                     ipakrbauthzdata=[u'MS-PAC'],
@@ -455,7 +457,7 @@ class test_service(Declarative):
                 summary=None,
                 result=dict(
                     dn=service1dn,
-                    usercertificate=[base64.b64decode(get_testcert())],
+                    usercertificate=[base64.b64decode(servercert)],
                     krbprincipalname=[service1],
                     has_keytab=False,
                     managedby_host=[fqdn1],
@@ -482,7 +484,7 @@ class test_service(Declarative):
                 value=service1,
                 summary=u'Modified service "%s"' % service1,
                 result=dict(
-                    usercertificate=[base64.b64decode(get_testcert())],
+                    usercertificate=[base64.b64decode(servercert)],
                     krbprincipalname=[service1],
                     managedby_host=[fqdn1],
                     ipakrbauthzdata=[u'MS-PAC'],
@@ -509,7 +511,7 @@ class test_service(Declarative):
                 value=service1,
                 summary=u'Modified service "%s"' % service1,
                 result=dict(
-                    usercertificate=[base64.b64decode(get_testcert())],
+                    usercertificate=[base64.b64decode(servercert)],
                     krbprincipalname=[service1],
                     managedby_host=[fqdn1],
                     ipakrbauthzdata=[u'MS-PAC'],
@@ -534,7 +536,7 @@ class test_service(Declarative):
                 value=service1,
                 summary=u'Modified service "%s"' % service1,
                 result=dict(
-                    usercertificate=[base64.b64decode(get_testcert())],
+                    usercertificate=[base64.b64decode(servercert)],
                     krbprincipalname=[service1],
                     managedby_host=[fqdn1],
                     ipakrbauthzdata=[u'MS-PAC'],
@@ -574,7 +576,7 @@ class test_service(Declarative):
 
         dict(
             desc='Try to update non-existent %r' % service1,
-            command=('service_mod', [service1], dict(usercertificate=get_testcert())),
+            command=('service_mod', [service1], dict(usercertificate=servercert)),
             expected=errors.NotFound(
                 reason=u'%s: service not found' % service1),
         ),
diff --git a/ipatests/test_xmlrpc/testcert.py b/ipatests/test_xmlrpc/testcert.py
index 77179fc78..4afd38d2c 100644
--- a/ipatests/test_xmlrpc/testcert.py
+++ b/ipatests/test_xmlrpc/testcert.py
@@ -34,18 +34,15 @@ from ipapython import ipautil
 from ipaplatform.paths import paths
 from ipapython.dn import DN
 
-_testcert = None
 
-
-def get_testcert():
+def get_testcert(subject, principal):
     """Get the certificate, creating it if it doesn't exist"""
-    global _testcert
-    if _testcert is None:
-        reqdir = tempfile.mkdtemp(prefix="tmp-")
-        try:
-            _testcert = makecert(reqdir)
-        finally:
-            shutil.rmtree(reqdir)
+    reqdir = tempfile.mkdtemp(prefix="tmp-")
+    try:
+        _testcert = makecert(reqdir, subject,
+                             principal)
+    finally:
+        shutil.rmtree(reqdir)
     return x509.strip_header(_testcert)
 
 
@@ -72,9 +69,9 @@ def generate_csr(reqdir, pwname, subject):
         return fp.read()
 
 
-def makecert(reqdir):
+def makecert(reqdir, subject, principal):
     """
-    Generate a service certificate that can be used during unit testing.
+    Generate a certificate that can be used during unit testing.
     """
 
     ra = rabase.rabase()
@@ -96,9 +93,7 @@ def makecert(reqdir):
     subject_base = res['result']['ipacertificatesubjectbase'][0]
 
     cert = None
-    subject = DN(('CN', api.env.host), subject_base)
-    princ = 'unittest/%s@%s' % (api.env.host, api.env.realm)
     csr = unicode(generate_csr(reqdir, pwname, str(subject)))
 
-    res = api.Command['cert_request'](csr, principal=princ, add=True)
+    res = api.Command['cert_request'](csr, principal=principal, add=True)
     return x509.make_pem(res['result']['certificate'])