Synnefo » snf-ganeti

  hyper.PowercycleNode()

  """Verifies a remote command name.

  @type cmd: string

  return (True, None)

  """Common checks for remote command file system directories and files.

  @type path: string

  return (True, st)

  """Verifies remote command directory.

  @type path: string

    element is an error message string, otherwise it's C{None}

  """

  if not status:

    return (False, value)

  return (True, None)

  """Verifies a whole remote command and returns its executable filename.

  @type path: string

  """

  executable = utils.PathJoin(path, cmd)

  if not status:

    return (False, msg)

  return (True, executable)

  """Performs a number of tests on a remote command.

  @type path: string

  @param path: Directory containing remote commands

  @type cmd: string

  @param cmd: Command name

  """

  # Verify the directory first

                     _lock_file=pathutils.RESTRICTED_COMMANDS_LOCK_FILE,

                     _path=pathutils.RESTRICTED_COMMANDS_DIR,

                     _sleep_fn=time.sleep,

                     _runcmd_fn=utils.RunCmd,

                     _enabled=constants.ENABLE_RESTRICTED_COMMANDS):

  """Executes a remote command after performing strict tests.

    tmpname = self._PrepareTest()

    os.chown(tmpname, 0, 0)

    self.assertTrue(status)

    self.assertTrue(value)

      self.assertFalse(uid == os.getuid() and gid == os.getgid())

      os.chown(tmpname, uid, gid)

      self.assertFalse(status)

      self.assertTrue("foobar' is not owned by " in errmsg)

                "Result from netutils.TcpPing corrupted")

  return (os.getuid(), os.getgid())

  def testAcceptableName(self):

    for i in ["foo", "bar", "z1", "000first", "hello-world"]:

      for fn in [lambda s: s, lambda s: s.upper(), lambda s: s.title()]:

        self.assertTrue(status)

        self.assertTrue(msg is None)

  def testEmptyAndSpace(self):

    for i in ["", " ", "\t", "\n"]:

      self.assertFalse(status)

      self.assertEqual(msg, "Missing command name")

  def testNameWithSlashes(self):

    for i in ["/", "./foo", "../moo", "some/name"]:

      self.assertFalse(status)

      self.assertEqual(msg, "Invalid command name")

  def testForbiddenCharacters(self):

    for i in ["#", ".", "..", "bash -c ls", "'"]:

      self.assertFalse(status)

      self.assertEqual(msg, "Command name contains forbidden characters")

  def setUp(self):

    self.tmpdir = tempfile.mkdtemp()

    tmpname = utils.PathJoin(self.tmpdir, "foobar")

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

    (status, msg) = \

    self.assertFalse(status)

    self.assertTrue(msg.startswith("Can't stat(2) '"))

      os.chmod(tmpname, mode)

      self.assertTrue(os.path.isdir(tmpname))

      (status, msg) = \

      self.assertFalse(status)

      self.assertTrue(msg.startswith("Permissions on '"))

    utils.WriteFile(tmpname, data="empty\n")

    self.assertTrue(os.path.isfile(tmpname))

    (status, msg) = \

    self.assertFalse(status)

    self.assertTrue(msg.endswith("is not a directory"))

    os.mkdir(tmpname)

    self.assertTrue(os.path.isdir(tmpname))

    (status, msg) = \

    self.assertTrue(status)

    self.assertTrue(msg is None)

  def setUp(self):

    self.tmpdir = tempfile.mkdtemp()

    tmpname = utils.PathJoin(self.tmpdir, "helloworld")

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

    (status, msg) = \

                                   _owner=NotImplemented)

    self.assertFalse(status)

    self.assertTrue(msg.startswith("Can't stat(2) '"))

    tmpname = utils.PathJoin(self.tmpdir, "cmdname")

    utils.WriteFile(tmpname, data="empty\n")

    (status, msg) = \

    self.assertFalse(status)

    self.assertTrue(msg.startswith("access(2) thinks '"))

    tmpname = utils.PathJoin(self.tmpdir, "cmdname")

    utils.WriteFile(tmpname, data="empty\n", mode=0700)

    (status, executable) = \

    self.assertTrue(status)

    self.assertEqual(executable, tmpname)

  _TEST_PATH = "/tmp/some/test/path"

  def testDirFails(self):

      return (False, "test error 31420")

    (status, msg) = \

                                    _verify_dir=fn,

                                    _verify_name=NotImplemented,

                                    _verify_cmd=NotImplemented)

      return (False, "test error 591")

    (status, msg) = \

                                    _verify_dir=lambda _: (True, None),

                                    _verify_name=fn,

                                    _verify_cmd=NotImplemented)

      return (False, "test error 25524")

    (status, msg) = \

                                    _verify_dir=lambda _: (True, None),

                                    _verify_name=lambda _: (True, None),

                                    _verify_cmd=fn)

      return (True, utils.PathJoin(path, cmd))

    (status, executable) = \

                                    _verify_dir=lambda _: (True, None),

                                    _verify_name=lambda _: (True, None),

                                    _verify_cmd=fn)

    self.assertEqual(executable, utils.PathJoin(self._TEST_PATH, "cmd22633"))

  assert duration > 5

  return "Executing command '%s' failed" % cmd

  def testNonExistantLockDirectory(self):

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

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

    self.assertRaises(backend.RPCFail,

                      backend.RunRestrictedCmd, "test",

  @staticmethod

  def _TryLock(lockfile):

    result = False

    try:

                               _runcmd_fn=NotImplemented,

                               _enabled=True)

    except backend.RPCFail, err:

             "Did not fail with generic error message"

      result = True

  def testPrepareRaisesException(self):

    lockfile = utils.PathJoin(self.tmpdir, "lock")

    prepare_fn = testutils.CallCounter(self._PrepareRaisingException)

    try:

                               _sleep_fn=sleep_fn, _prepare_fn=prepare_fn,

                               _enabled=True)

    except backend.RPCFail, err:

    else:

      self.fail("Didn't fail")

  def testPrepareFails(self):

    lockfile = utils.PathJoin(self.tmpdir, "lock")

    prepare_fn = testutils.CallCounter(self._PrepareFails)

    try:

                               _sleep_fn=sleep_fn, _prepare_fn=prepare_fn,

                               _enabled=True)

    except backend.RPCFail, err:

    else:

      self.fail("Didn't fail")

                             utils.ShellQuoteArgs(args),

                             NotImplemented, NotImplemented)

    prepare_fn = testutils.CallCounter(self._SuccessfulPrepare)

    runcmd_fn = testutils.CallCounter(fn)

                             utils.ShellQuoteArgs(args),

                             NotImplemented, NotImplemented)

    prepare_fn = testutils.CallCounter(self._SuccessfulPrepare)

    runcmd_fn = testutils.CallCounter(fn)