Synnefo » snf-ganeti

class TestCheckNodeCertificate(testutils.GanetiTestCase):

  def setUp(self):

    testutils.GanetiTestCase.setUp(self)

    self.tmpdir = tempfile.mkdtemp()

  def tearDown(self):

    testutils.GanetiTestCase.tearDown(self)

    shutil.rmtree(self.tmpdir)

  def testMismatchingKey(self):

    other_cert = self._TestDataFilename("cert1.pem")

    node_cert = self._TestDataFilename("cert2.pem")

    cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,

                                           utils.ReadFile(other_cert))

    try:

      utils.CheckNodeCertificate(cert, _noded_cert_file=node_cert)

    except errors.GenericError, err:

      self.assertEqual(str(err),

                       "Given cluster certificate does not match local key")

    else:

      self.fail("Exception was not raised")

  def testMatchingKey(self):

    cert_filename = self._TestDataFilename("cert2.pem")

    # Extract certificate

    cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,

                                           utils.ReadFile(cert_filename))

    cert_pem = OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_PEM,

                                               cert)

    utils.CheckNodeCertificate(cert, _noded_cert_file=cert_filename)

  def testMissingFile(self):

    cert_path = self._TestDataFilename("cert1.pem")

    nodecert = utils.PathJoin(self.tmpdir, "does-not-exist")

    utils.CheckNodeCertificate(NotImplemented, _noded_cert_file=nodecert)

    self.assertFalse(os.path.exists(nodecert))

  def testInvalidCertificate(self):

    tmpfile = utils.PathJoin(self.tmpdir, "cert")

    utils.WriteFile(tmpfile, data="not a certificate")

    self.assertRaises(errors.X509CertError, utils.CheckNodeCertificate,

                      NotImplemented, _noded_cert_file=tmpfile)

  def testNoPrivateKey(self):

    cert = self._TestDataFilename("cert1.pem")

    self.assertRaises(errors.X509CertError, utils.CheckNodeCertificate,

                      NotImplemented, _noded_cert_file=cert)

  def testMismatchInNodeCert(self):

    cert1_path = self._TestDataFilename("cert1.pem")

    cert2_path = self._TestDataFilename("cert2.pem")

    tmpfile = utils.PathJoin(self.tmpdir, "cert")

    # Extract certificate

    cert1 = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,

                                            utils.ReadFile(cert1_path))

    cert1_pem = OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_PEM,

                                                cert1)

    # Extract mismatching key

    key2 = OpenSSL.crypto.load_privatekey(OpenSSL.crypto.FILETYPE_PEM,

                                          utils.ReadFile(cert2_path))

    key2_pem = OpenSSL.crypto.dump_privatekey(OpenSSL.crypto.FILETYPE_PEM,

                                              key2)

    # Write to file

    utils.WriteFile(tmpfile, data=cert1_pem + key2_pem)

    try:

      utils.CheckNodeCertificate(cert1, _noded_cert_file=tmpfile)

    except errors.X509CertError, err:

      self.assertEqual(err.args,

                       (tmpfile, "Certificate does not match with private key"))

    else:

      self.fail("Exception was not raised")