Synnefo » snf-ganeti

	test/ganeti.backend_unittest.py \

  # Crypto

  @staticmethod

  def perspective_create_x509_certificate(params):

    """Creates a new X509 certificate for SSL/TLS.

    """

    (validity, ) = params

    return backend.CreateX509Certificate(validity)

  @staticmethod

  def perspective_remove_x509_certificate(params):

    """Removes a X509 certificate.

    """

    (name, ) = params

    return backend.RemoveX509Certificate(name)

  dirs.append((constants.CRYPTO_KEYS_DIR, constants.CRYPTO_KEYS_DIR_MODE))

  constants.CRYPTO_KEYS_DIR,

_MAX_SSL_CERT_VALIDITY = 7 * 24 * 60 * 60

_X509_KEY_FILE = "key"

_X509_CERT_FILE = "cert"

  _CleanDirectory(constants.CRYPTO_KEYS_DIR)

def _GetX509Filenames(cryptodir, name):

  """Returns the full paths for the private key and certificate.

  """

  return (utils.PathJoin(cryptodir, name),

          utils.PathJoin(cryptodir, name, _X509_KEY_FILE),

          utils.PathJoin(cryptodir, name, _X509_CERT_FILE))

def CreateX509Certificate(validity, cryptodir=constants.CRYPTO_KEYS_DIR):

  """Creates a new X509 certificate for SSL/TLS.

  @type validity: int

  @param validity: Validity in seconds

  @rtype: tuple; (string, string)

  @return: Certificate name and public part

  """

  (key_pem, cert_pem) = \

    utils.GenerateSelfSignedX509Cert(utils.HostInfo.SysName(),

                                     min(validity, _MAX_SSL_CERT_VALIDITY))

  cert_dir = tempfile.mkdtemp(dir=cryptodir,

                              prefix="x509-%s-" % utils.TimestampForFilename())

  try:

    name = os.path.basename(cert_dir)

    assert len(name) > 5

    (_, key_file, cert_file) = _GetX509Filenames(cryptodir, name)

    utils.WriteFile(key_file, mode=0400, data=key_pem)

    utils.WriteFile(cert_file, mode=0400, data=cert_pem)

    # Never return private key as it shouldn't leave the node

    return (name, cert_pem)

  except Exception:

    shutil.rmtree(cert_dir, ignore_errors=True)

    raise

def RemoveX509Certificate(name, cryptodir=constants.CRYPTO_KEYS_DIR):

  """Removes a X509 certificate.

  @type name: string

  @param name: Certificate name

  """

  (cert_dir, key_file, cert_file) = _GetX509Filenames(cryptodir, name)

  utils.RemoveFile(key_file)

  utils.RemoveFile(cert_file)

  try:

    os.rmdir(cert_dir)

  except EnvironmentError, err:

    _Fail("Cannot remove certificate directory '%s': %s",

          cert_dir, err)

CRYPTO_KEYS_DIR = RUN_GANETI_DIR + "/crypto"

CRYPTO_KEYS_DIR_MODE = 0700

  def call_create_x509_certificate(self, node, validity):

    """Creates a new X509 certificate for SSL/TLS.

    This is a single-node call.

    @type validity: int

    @param validity: Validity in seconds

    """

    return self._SingleNodeCall(node, "create_x509_certificate", [validity])

  def call_remove_x509_certificate(self, node, name):

    """Removes a X509 certificate.

    This is a single-node call.

    @type name: string

    @param name: Certificate name

    """

    return self._SingleNodeCall(node, "remove_x509_certificate", [name])

#!/usr/bin/python

#

# Copyright (C) 2010 Google Inc.

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful, but

# WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

# General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

# 02110-1301, USA.

"""Script for testing ganeti.backend"""

import os

import sys

import shutil

import tempfile

import unittest

from ganeti import utils

from ganeti import backend

import testutils

class TestX509Certificates(unittest.TestCase):

  def setUp(self):

    self.tmpdir = tempfile.mkdtemp()

  def tearDown(self):

    shutil.rmtree(self.tmpdir)

  def test(self):

    (name, cert_pem) = backend.CreateX509Certificate(300, cryptodir=self.tmpdir)

    self.assertEqual(utils.ReadFile(os.path.join(self.tmpdir, name,

                                                 backend._X509_CERT_FILE)),

                     cert_pem)

    self.assert_(0 < os.path.getsize(os.path.join(self.tmpdir, name,

                                                  backend._X509_KEY_FILE)))

    (name2, cert_pem2) = \

      backend.CreateX509Certificate(300, cryptodir=self.tmpdir)

    backend.RemoveX509Certificate(name, cryptodir=self.tmpdir)

    backend.RemoveX509Certificate(name2, cryptodir=self.tmpdir)

    self.assertEqual(utils.ListVisibleFiles(self.tmpdir), [])

  def testNonEmpty(self):

    (name, _) = backend.CreateX509Certificate(300, cryptodir=self.tmpdir)

    utils.WriteFile(utils.PathJoin(self.tmpdir, name, "hello-world"),

                    data="Hello World")

    self.assertRaises(backend.RPCFail, backend.RemoveX509Certificate,

                      name, cryptodir=self.tmpdir)

    self.assertEqual(utils.ListVisibleFiles(self.tmpdir), [name])

if __name__ == "__main__":

  testutils.GanetiTestProgram()