Synnefo » snf-ganeti

import OpenSSL

def _FormatTimestamp(secs):

  """Formats a Unix timestamp with the local timezone.

  """

  return time.strftime("%F %T %Z", time.gmtime(secs))

def _VerifyCertificateInner(filename, expired, not_before, not_after, now,

                            warn_days=constants.SSL_CERT_EXPIRATION_WARN,

                            error_days=constants.SSL_CERT_EXPIRATION_ERROR):

  """Verifies certificate details for LUVerifyCluster.

  """

  if expired:

    msg = "Certificate %s is expired" % filename

    if not_before is not None and not_after is not None:

      msg += (" (valid from %s to %s)" %

              (_FormatTimestamp(not_before),

               _FormatTimestamp(not_after)))

    elif not_before is not None:

      msg += " (valid from %s)" % _FormatTimestamp(not_before)

    elif not_after is not None:

      msg += " (valid until %s)" % _FormatTimestamp(not_after)

    return (LUVerifyCluster.ETYPE_ERROR, msg)

  elif not_before is not None and not_before > now:

    return (LUVerifyCluster.ETYPE_WARNING,

            "Certificate %s not yet valid (valid from %s)" %

            (filename, _FormatTimestamp(not_before)))

  elif not_after is not None:

    remaining_days = int((not_after - now) / (24 * 3600))

    msg = ("Certificate %s expires in %d days" % (filename, remaining_days))

    if remaining_days <= error_days:

      return (LUVerifyCluster.ETYPE_ERROR, msg)

    if remaining_days <= warn_days:

      return (LUVerifyCluster.ETYPE_WARNING, msg)

  return (None, None)

def _VerifyCertificate(filename):

  """Verifies a certificate for LUVerifyCluster.

  @type filename: string

  @param filename: Path to PEM file

  """

  try:

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

                                           utils.ReadFile(filename))

  except Exception, err: # pylint: disable-msg=W0703

    return (LUVerifyCluster.ETYPE_ERROR,

            "Failed to load X509 certificate %s: %s" % (filename, err))

  # Depending on the pyOpenSSL version, this can just return (None, None)

  (not_before, not_after) = utils.GetX509CertValidity(cert)

  return _VerifyCertificateInner(filename, cert.has_expired(),

                                 not_before, not_after, time.time())

  ECLUSTERCERT = (TCLUSTER, "ECLUSTERCERT")

    # Check the cluster certificates

    for cert_filename in constants.ALL_CERT_FILES:

      (errcode, msg) = _VerifyCertificate(cert_filename)

      _ErrorIf(errcode, self.ECLUSTERCERT, None, msg, code=errcode)

# SSL certificate check constants (in days)

SSL_CERT_EXPIRATION_WARN = 30

SSL_CERT_EXPIRATION_ERROR = 7

import tempfile

import shutil

class TestCertVerification(testutils.GanetiTestCase):

  def setUp(self):

    testutils.GanetiTestCase.setUp(self)

    self.tmpdir = tempfile.mkdtemp()

  def tearDown(self):

    shutil.rmtree(self.tmpdir)

  def testVerifyCertificate(self):

    cmdlib._VerifyCertificate(self._TestDataFilename("cert1.pem"))

    nonexist_filename = os.path.join(self.tmpdir, "does-not-exist")

    (errcode, msg) = cmdlib._VerifyCertificate(nonexist_filename)

    self.assertEqual(errcode, cmdlib.LUVerifyCluster.ETYPE_ERROR)

    # Try to load non-certificate file

    invalid_cert = self._TestDataFilename("bdev-net1.txt")

    (errcode, msg) = cmdlib._VerifyCertificate(invalid_cert)

    self.assertEqual(errcode, cmdlib.LUVerifyCluster.ETYPE_ERROR)

class TestVerifyCertificateInner(unittest.TestCase):

  FAKEFILE = "/tmp/fake/cert/file.pem"

  def test(self):

    vci = cmdlib._VerifyCertificateInner

    # Valid

    self.assertEqual(vci(self.FAKEFILE, False, 1263916313, 1298476313,

                         1266940313, warn_days=30, error_days=7),

                     (None, None))

    # Not yet valid

    (errcode, msg) = vci(self.FAKEFILE, False, 1266507600, 1267544400,

                         1266075600, warn_days=30, error_days=7)

    self.assertEqual(errcode, cmdlib.LUVerifyCluster.ETYPE_WARNING)

    # Expiring soon

    (errcode, msg) = vci(self.FAKEFILE, False, 1266507600, 1267544400,

                         1266939600, warn_days=30, error_days=7)

    self.assertEqual(errcode, cmdlib.LUVerifyCluster.ETYPE_ERROR)

    (errcode, msg) = vci(self.FAKEFILE, False, 1266507600, 1267544400,

                         1266939600, warn_days=30, error_days=1)

    self.assertEqual(errcode, cmdlib.LUVerifyCluster.ETYPE_WARNING)

    (errcode, msg) = vci(self.FAKEFILE, False, 1266507600, None,

                         1266939600, warn_days=30, error_days=7)

    self.assertEqual(errcode, None)

    # Expired

    (errcode, msg) = vci(self.FAKEFILE, True, 1266507600, 1267544400,

                         1266939600, warn_days=30, error_days=7)

    self.assertEqual(errcode, cmdlib.LUVerifyCluster.ETYPE_ERROR)

    (errcode, msg) = vci(self.FAKEFILE, True, None, 1267544400,

                         1266939600, warn_days=30, error_days=7)

    self.assertEqual(errcode, cmdlib.LUVerifyCluster.ETYPE_ERROR)

    (errcode, msg) = vci(self.FAKEFILE, True, 1266507600, None,

                         1266939600, warn_days=30, error_days=7)

    self.assertEqual(errcode, cmdlib.LUVerifyCluster.ETYPE_ERROR)

    (errcode, msg) = vci(self.FAKEFILE, True, None, None,

                         1266939600, warn_days=30, error_days=7)

    self.assertEqual(errcode, cmdlib.LUVerifyCluster.ETYPE_ERROR)

if __name__ == "__main__":

  def testSslCertExpiration(self):

    self.failUnless(constants.SSL_CERT_EXPIRATION_ERROR <

                    constants.SSL_CERT_EXPIRATION_WARN)