erp5_certificate_authority: Update test for Person/Certificate Login

bt5/erp5_certificate_authority/MixinTemplateItem/portal_components/mixin.erp5.CertificateLoginMixin.py

@@ -72,9 +72,11 @@ class CertificateLoginMixin:
     certificate_dict = {
       "common_name" : self.getReference()
     }
-    if self.getReference and self.getSourceReference():
+    if self.getReference() and self.getSourceReference():
+      if csr is not None:
+        raise ValueError("This certificate already")
       certificate_dict["id"] = self.getSourceReference()
-      crt_pem = caucase_connector.getCertificate(self.getSourceReference())
+      crt_pem = caucase_connector.getCertificate(int(self.getSourceReference()))
       certificate_dict["certificate"] = crt_pem
       # We should assert that reference is the CN of crt_pem
       return certificate_dict
@@ -111,7 +113,7 @@ class CertificateLoginMixin:
     else:
       return self._getCertificate(csr=csr)
 
-  def _revokeCertificate(self):
+  def _revokeCertificate(self, key_pem=None):
     if self.getDestinationReference() is not None or (
       self.getReference() is not None and self.getSourceReference() is None
     ):
@@ -119,12 +121,14 @@ class CertificateLoginMixin:
     
     if self.getSourceReference() is not None:
       caucase_connector = self._getCaucaseConnector()
-      crt_pem = caucase_connector.getCertificate(self.getSourceReference())
-      caucase_connector.revokeCertificate(crt_pem)
+      crt_pem = caucase_connector.getCertificate(int(self.getSourceReference()))
+      if key_pem is None:
+        return caucase_connector.revokeCertificate(crt_pem, key_pem)
+      return caucase_connector.revokeCertificate(crt_pem)
     else:
       raise ValueError("No certificate found to revoke!")
 
   security.declarePrivate('revokeCertificate')
-  def revokeCertificate(self):
+  def revokeCertificate(self, key_pem=None):
     """Revokes existing certificate"""
-    self._revokeCertificate()
+    self._revokeCertificate(key_pem=key_pem)
\ No newline at end of file

bt5/erp5_certificate_authority/TestTemplateItem/portal_components/test.erp5.testCertificateAuthorityPerson.py

@@ -27,15 +27,20 @@
 #
 ##############################################################################
 
-import os
+#import os
 import random
-from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
+from Products.ERP5Type.tests.ERP5TypeCaucaseTestCase import ERP5TypeCaucaseTestCase
 from Products.ERP5Type.Core.Workflow import ValidationFailed
 from AccessControl import Unauthorized
+from caucase.client import CaucaseHTTPError
 
-class TestPersonCertificateLogin(ERP5TypeTestCase):
+from cryptography import x509
+from cryptography.x509.oid import NameOID
+
+class TestPersonCertificateLogin(ERP5TypeCaucaseTestCase):
 
   def afterSetUp(self):
+    self.setUpCaucase()
     if getattr(self.portal.portal_types.Person,
         'user_can_see_himself', None) is None:
       self.portal.portal_types.Person.newContent(
@@ -45,9 +50,6 @@ class TestPersonCertificateLogin(ERP5TypeTestCase):
             role_base_category_script_id="ERP5Type_getSecurityCategoryFromSelf",
             role_base_category="group",
             portal_type="Role Information")
-    if "TEST_CA_PATH" in os.environ:
-      self.portal.portal_certificate_authority.certificate_authority_path = \
-          os.environ['TEST_CA_PATH']
 
   def getBusinessTemplateList(self):
     return ('erp5_base', 'erp5_certificate_authority')
@@ -76,8 +78,10 @@ class TestPersonCertificateLogin(ERP5TypeTestCase):
     self.assertNotEqual(certificate_login.getReference(), login)
     self.assertTrue(certificate_login.getReference().startswith("CERT"))
 
-    self.assertIn('CN=%s' % certificate_login.getReference(), certificate['certificate'])
-    self.assertNotIn('CN=%s' % user_id, certificate['certificate'])
+    ssl_certificate = x509.load_pem_x509_certificate(certificate['certificate'])
+    self.assertEqual(len(ssl_certificate.subject), 1)
+    cn = [i.value for i in ssl_certificate.subject if i.oid == NameOID.COMMON_NAME][0]
+    self.assertEqual(certificate_login.getReference().decode("UTF-8"), cn)
 
   def test_person_duplicated_login(self):
     user_id, login = self._createPerson()
@@ -96,8 +100,11 @@ class TestPersonCertificateLogin(ERP5TypeTestCase):
     self.assertNotEqual(certificate_login.getReference(), user_id)
     self.assertNotEqual(certificate_login.getReference(), login)
     self.assertTrue(certificate_login.getReference().startswith("CERT"))
-    self.assertIn('CN=%s' % certificate_login.getReference(), certificate['certificate'])
-    self.assertNotIn('CN=%s' % user_id, certificate['certificate'])
+
+    ssl_certificate = x509.load_pem_x509_certificate(certificate['certificate'])
+    self.assertEqual(len(ssl_certificate.subject), 1)
+    cn = [i.value for i in ssl_certificate.subject if i.oid == NameOID.COMMON_NAME][0]
+    self.assertEqual(certificate_login.getReference().decode("UTF-8"), cn)
 
     # ERP5 Login dont conflicts
     person.newContent(portal_type='ERP5 Login',
@@ -117,10 +124,12 @@ class TestPersonCertificateLogin(ERP5TypeTestCase):
     self.assertNotEqual(certificate_login.getReference(), user_id)
     self.assertNotEqual(certificate_login.getReference(), login)
     self.assertTrue(certificate_login.getReference().startswith("CERT"))
-    
-    self.assertIn('CN=%s' % certificate_login.getReference(), certificate['certificate'])
-    self.assertNotIn('CN=%s' % user_id, certificate['certificate'])
-    self.assertNotIn('CN=%s' % login, certificate['certificate'])
+
+    ssl_certificate = x509.load_pem_x509_certificate(certificate['certificate'])
+    self.assertEqual(len(ssl_certificate.subject), 1)
+    cn = [i.value for i in ssl_certificate.subject if i.oid == NameOID.COMMON_NAME][0]
+    self.assertEqual(certificate_login.getReference().decode("UTF-8"), cn)
+
     self.assertEqual(certificate_login.getValidationState(), "validated")
 
     new_certificate = person.generateCertificate()
@@ -139,11 +148,12 @@ class TestPersonCertificateLogin(ERP5TypeTestCase):
       certificate_login.getReference())
     
     self.assertTrue(new_certificate_login.getReference().startswith("CERT"))
-    
-    self.assertIn('CN=%s' % new_certificate_login.getReference(), new_certificate['certificate'])
-    self.assertNotIn('CN=%s' % user_id, new_certificate['certificate'])
-    self.assertNotIn('CN=%s' % login, new_certificate['certificate'])
-    self.assertNotIn('CN=%s' % certificate_login.getReference(), new_certificate['certificate'])
+
+    ssl_certificate = x509.load_pem_x509_certificate(new_certificate['certificate'])
+    self.assertEqual(len(ssl_certificate.subject), 1)
+    cn = [i.value for i in ssl_certificate.subject if i.oid == NameOID.COMMON_NAME][0]
+    self.assertEqual(new_certificate_login.getReference().decode("UTF-8"), cn)
+
     self.assertEqual(new_certificate_login.getValidationState(), "validated")
 
   def test_person_generate_certificate_for_another(self):
@@ -191,9 +201,13 @@ class TestPersonCertificateLogin(ERP5TypeTestCase):
     self.assertNotEqual(certificate_login.getReference(), person.getUserId())
 
     self.assertTrue(certificate_login.getReference().startswith("CERT"))
-    
-    self.assertIn('CN=%s' % certificate_login.getReference(), certificate_dict['certificate'])
-    self.assertNotIn('CN=%s' % person.getUserId(), certificate_dict['certificate'])
+
+    ssl_certificate = x509.load_pem_x509_certificate(certificate_dict['certificate'])
+    self.assertEqual(len(ssl_certificate.subject), 1)
+    cn_list = [i.value for i in ssl_certificate.subject if i.oid == NameOID.COMMON_NAME]
+    self.assertEqual(len(cn_list), 1)
+    self.assertEqual(certificate_login.getReference().decode("UTF-8"), cn_list[0])
+
     self.assertEqual(certificate_login.getValidationState(), "draft")
 
     certificate_login.validate()
@@ -212,9 +226,12 @@ class TestPersonCertificateLogin(ERP5TypeTestCase):
 
     # Reference is reset while setting the generate the certificate.
     self.assertTrue(certificate_login.getReference().startswith("CERT"))
-    
-    self.assertIn('CN=%s' % certificate_login.getReference(), certificate_dict['certificate'])
-    self.assertNotIn('CN=%s' % person.getUserId(), certificate_dict['certificate'])
+
+    ssl_certificate = x509.load_pem_x509_certificate(certificate_dict['certificate'])
+    self.assertEqual(len(ssl_certificate.subject), 1)
+    cn_list = [i.value for i in ssl_certificate.subject if i.oid == NameOID.COMMON_NAME]
+    self.assertEqual(len(cn_list), 1)
+    self.assertEqual(certificate_login.getReference().decode("UTF-8"), cn_list[0])
     self.assertEqual(certificate_login.getValidationState(), "draft")
 
     certificate_login.validate()
@@ -230,12 +247,26 @@ class TestPersonCertificateLogin(ERP5TypeTestCase):
     reference = certificate_login.getReference()
     # Reference is reset while setting the generate the certificate.
     self.assertTrue(reference.startswith("CERT"))
-    
-    self.assertIn('CN=%s' % reference, certificate_dict['certificate'])
-    self.assertNotIn('CN=%s' % person.getUserId(), certificate_dict['certificate'])
+
+    # If no csr is provided, the private key is generated by the master
+    # this is to provide backward compatibility with old clients
+    self.assertIn("key", certificate_dict.keys())
+
+    ssl_certificate = x509.load_pem_x509_certificate(certificate_dict['certificate'])
+    self.assertEqual(len(ssl_certificate.subject), 1)
+    cn_list = [i.value for i in ssl_certificate.subject if i.oid == NameOID.COMMON_NAME]
+    self.assertEqual(len(cn_list), 1)
+    self.assertEqual(certificate_login.getReference().decode("UTF-8"), cn_list[0])
     self.assertEqual(certificate_login.getValidationState(), "draft")
 
-    self.assertRaises(ValueError, certificate_login.getCertificate)
+    same_certificate_dict = certificate_login.getCertificate()
+    self.assertEqual(certificate_dict['certificate'], same_certificate_dict['certificate'])
+    
+    # If no csr is provided, the private key is generated by the master
+    # this is to provide backward compatibility with old clients
+    self.assertNotIn("key", same_certificate_dict.keys())
+
+    self.assertRaises(ValueError, certificate_login.getCertificate, "some_csr_string")
 
   def test_certificate_login_revoke(self):
     person = self.portal.person_module.newContent(portal_type='Person')
@@ -246,15 +277,48 @@ class TestPersonCertificateLogin(ERP5TypeTestCase):
     certificate_dict = certificate_login.getCertificate()
     reference = certificate_login.getReference()
     self.assertTrue(reference.startswith("CERT"))
-    self.assertIn('CN=%s' % reference, certificate_dict['certificate'])
-    self.assertNotEqual(certificate_login.getDestinationReference(), None)
+    
+    ssl_certificate = x509.load_pem_x509_certificate(certificate_dict['certificate'])
+    self.assertEqual(len(ssl_certificate.subject), 1)
+    cn_list = [i.value for i in ssl_certificate.subject if i.oid == NameOID.COMMON_NAME]
+    self.assertEqual(len(cn_list), 1)
+    self.assertEqual(certificate_login.getReference().decode("UTF-8"), cn_list[0])
+
+    self.assertEqual(certificate_login.getDestinationReference(), None)
+    self.assertNotEqual(certificate_login.getSourceReference(), None)
 
     self.assertEqual(None, certificate_login.revokeCertificate())
     self.assertEqual(certificate_login.getDestinationReference(), None)
     self.assertEqual(reference, certificate_login.getReference())
 
     # Revoke again must raise
+    self.assertRaises(CaucaseHTTPError, certificate_login.revokeCertificate)
+
+  def test_certificate_login_revoke_providing_key(self):
+    person = self.portal.person_module.newContent(portal_type='Person')
+    certificate_login = person.newContent(portal_type='Certificate Login')
+    self.assertEqual(certificate_login.getReference(), None)
     self.assertRaises(ValueError, certificate_login.revokeCertificate)
+    
+    certificate_dict = certificate_login.getCertificate()
+    reference = certificate_login.getReference()
+    self.assertTrue(reference.startswith("CERT"))
+    
+    ssl_certificate = x509.load_pem_x509_certificate(certificate_dict['certificate'])
+    self.assertEqual(len(ssl_certificate.subject), 1)
+    cn_list = [i.value for i in ssl_certificate.subject if i.oid == NameOID.COMMON_NAME]
+    self.assertEqual(len(cn_list), 1)
+    self.assertEqual(certificate_login.getReference().decode("UTF-8"), cn_list[0])
+
+    self.assertEqual(certificate_login.getDestinationReference(), None)
+    self.assertNotEqual(certificate_login.getSourceReference(), None)
+
+    self.assertEqual(None, certificate_login.revokeCertificate(certificate_dict['key']))
+    self.assertEqual(certificate_login.getDestinationReference(), None)
+    self.assertEqual(reference, certificate_login.getReference())
+
+    # Revoke again must raise
+    self.assertRaises(CaucaseHTTPError, certificate_login.revokeCertificate, certificate_dict['key'])
 
   def test_certificate_login_revoke_backward_compatibility(self):
     person = self.portal.person_module.newContent(portal_type='Person')
@@ -265,14 +329,36 @@ class TestPersonCertificateLogin(ERP5TypeTestCase):
     certificate_dict = certificate_login.getCertificate()
     reference = certificate_login.getReference()
     self.assertTrue(reference.startswith("CERT"))
-    self.assertIn('CN=%s' % reference, certificate_dict['certificate'])
-    self.assertNotEqual(certificate_login.getDestinationReference(), None)
+
+    ssl_certificate = x509.load_pem_x509_certificate(certificate_dict['certificate'])
+    self.assertEqual(len(ssl_certificate.subject), 1)
+    cn_list = [i.value for i in ssl_certificate.subject if i.oid == NameOID.COMMON_NAME]
+    self.assertEqual(len(cn_list), 1)
+    self.assertEqual(certificate_login.getReference().decode("UTF-8"), cn_list[0])
+
+    self.assertEqual(certificate_login.getDestinationReference(), None)
+    self.assertNotEqual(certificate_login.getSourceReference(), None)
 
     # Older implementation wont set it on the Certificate login
     certificate_login.setDestinationReference(None)
-    self.assertEqual(None, certificate_login.revokeCertificate())
-    self.assertEqual(certificate_login.getDestinationReference(), None)
-    self.assertEqual(reference, certificate_login.getReference())
+    certificate_login.setSourceReference(None)
+
+    # Still raise since it has no valid certificate anymore
+    self.assertRaises(ValueError, certificate_login.revokeCertificate)
+
+  def test_certificate_login_revoke_backward_compatibility_with_old_serial(self):
+    person = self.portal.person_module.newContent(portal_type='Person')
+    certificate_login = person.newContent(portal_type='Certificate Login')
+    self.assertEqual(certificate_login.getReference(), None)
+    self.assertRaises(ValueError, certificate_login.revokeCertificate)
+    
+    certificate_login.getCertificate()
+    reference = certificate_login.getReference()
+    self.assertTrue(reference.startswith("CERT"))
+
+    # Older implementation, using openssl, would have destination reference set
+    # this just raise since it cannot be managed by caucase
+    certificate_login.setDestinationReference("SOMESERIAL")
 
     # Still raise since it has no valid certificate anymore
     self.assertRaises(ValueError, certificate_login.revokeCertificate)