Synnefo » snf-ganeti

  return serializer.DumpJson(request,

                             private_encoder=serializer.EncodeWithPrivateFields)

from ganeti import constants

def DumpJson(data, private_encoder=None):

  @param private_encoder: specify L{serializer.EncodeWithPrivateFields} if you

                          require the produced JSON to also contain private

                          parameters. Otherwise, they will encode to null.

  if private_encoder is None:

    # Do not leak private fields by default.

    private_encoder = EncodeWithoutPrivateFields

  encoded = simplejson.dumps(data, default=private_encoder)

  values = simplejson.loads(txt)

  # Hunt and seek for Private fields and wrap them.

  WrapPrivateValues(values)

  return values

def WrapPrivateValues(json):

  """Crawl a JSON decoded structure for private values and wrap them.

  @param json: the json-decoded value to protect.

  """

  # This function used to be recursive. I use this list to avoid actual

  # recursion, however, since this is a very high-traffic area.

  todo = [json]

  while todo:

    data = todo.pop()

    if isinstance(data, list): # Array

      for item in data:

        todo.append(item)

    elif isinstance(data, dict): # Object

      # This is kind of a kludge, but the only place where we know what should

      # be protected is in ganeti.opcodes, and not in a way that is helpful to

      # us, especially in such a high traffic method; on the other hand, the

      # Haskell `py_compat_fields` test should complain whenever this check

      # does not protect fields properly.

      for field in data:

        value = data[field]

        if field in constants.PRIVATE_PARAMETERS_BLACKLIST:

          if not field.endswith("_cluster"):

            data[field] = PrivateDict(value)

          else:

            for os in data[field]:

              value[os] = PrivateDict(value[os])

        else:

          todo.append(value)

    else: # Values

      pass

def DumpSignedJson(data, key, salt=None, key_selector=None,

                   private_encoder=None):

  @param private_encoder: see L{DumpJson}

  txt = DumpJson(data, private_encoder=private_encoder)

  WrapPrivateValues(signed_dict)

def EncodeWithoutPrivateFields(obj):

  if isinstance(obj, Private):

    return None

  raise TypeError(repr(obj) + " is not JSON serializable")

def EncodeWithPrivateFields(obj):

  if isinstance(obj, Private):

    return obj.Get()

  raise TypeError(repr(obj) + " is not JSON serializable")

-- | Parameters that should be protected

--

-- Python does not have a type system and can't automatically infer what should

-- be the resulting type of a JSON request. As a result, it must rely on this

-- list of parameter names to protect values correctly.

--

-- Names ending in _cluster will be treated as dicts of dicts of private values.

-- Otherwise they are considered dicts of private values.

privateParametersBlacklist :: [String]

privateParametersBlacklist = [ "osparams_private"

                             , "osparams_secret"

                             , "osparams_private_cluster"

                             ]

-- | Warn the user that the logging level is too low for production use.

debugModeConfidentialityWarning :: String

debugModeConfidentialityWarning =

  "ALERT: %s started in debug mode.\n\

  \ Private and secret parameters WILL be logged!\n"

               \print serializer.Dump(\

               \  encoded,\

               \  private_encoder=serializer.EncodeWithPrivateFields)"

               serialized

import doctest

from ganeti import objects

from ganeti import serializer

    {"1": 2,

     "foo": "bar"},

        "osparams:": serializer.PrivateDict({

                      "foo": 5,

                     })

      }

  ]

    _dump_fn = lambda data: dump_fn(

      data,

      private_encoder=serializer.EncodeWithPrivateFields

    )

      self.failUnless(_dump_fn(data).endswith("\n"))

      self.assertEqualValues(load_fn(_dump_fn(data)), data)

    _dump_fn = lambda *args, **kwargs: dump_fn(

      *args,

      private_encoder=serializer.EncodeWithPrivateFields,

      **kwargs

    )

      self.assertEqualValues(load_fn(_dump_fn(data, "mykey"), "mykey"),

      self.assertEqualValues(load_fn(_dump_fn(data, "myprivatekey",

                                              salt="mysalt"),

      self.assertEqualValues(load_fn(_dump_fn(data, "myprivatekey",

                                              salt="mysalt",

                                              key_selector="mykey_id"),

                        _dump_fn(data, "myprivatekey",

                                 salt="mysalt",

                                 key_selector="mykey_id"),

                      _dump_fn("test", "myprivatekey"),

class TestPrivate(unittest.TestCase):

  def testEquality(self):

    pDict = serializer.PrivateDict()

    pDict["bar"] = "egg"

    nDict = {"bar": "egg"}

    self.assertEqual(pDict, nDict, "PrivateDict-dict equality failure")

  def testPrivateDictUnprivate(self):

    pDict = serializer.PrivateDict()

    pDict["bar"] = "egg"

    uDict = pDict.Unprivate()

    nDict = {"bar": "egg"}

    self.assertEquals(type(uDict), dict,

                      "PrivateDict.Unprivate() did not return a dict")

    self.assertEqual(pDict, uDict, "PrivateDict.Unprivate() equality failure")

    self.assertEqual(nDict, uDict, "PrivateDict.Unprivate() failed to return")

  def testAttributeTransparency(self):

    class Dummy(object):

      pass

    dummy = Dummy()

    dummy.bar = "egg"

    pDummy = serializer.Private(dummy)

    self.assertEqual(pDummy.bar, "egg", "Failed to access attribute of Private")

  def testCallTransparency(self):

    foo = serializer.Private("egg")

    self.assertEqual(foo.upper(), "EGG", "Failed to call Private instance")

  def testFillDict(self):

    pDict = serializer.PrivateDict()

    pDict["bar"] = "egg"

    self.assertEqual(pDict, objects.FillDict({}, pDict))

  def testLeak(self):

    pDict = serializer.PrivateDict()

    pDict["bar"] = "egg"

    self.assertNotIn("egg", str(pDict), "Value leaked in str(PrivateDict)")

    self.assertNotIn("egg", repr(pDict), "Value leaked in repr(PrivateDict)")

    self.assertNotIn("egg", "{0}".format(pDict),

                     "Value leaked in PrivateDict.__format__")

    self.assertNotIn("egg", serializer.Dump(pDict),

                     "Value leaked in serializer.Dump(PrivateDict)")

  def testProperAccess(self):

    pDict = serializer.PrivateDict()

    pDict["bar"] = "egg"

    self.assertIs("egg", pDict["bar"].Get(),

                  "Value not returned by Private.Get()")

    self.assertIs("egg", pDict.GetPrivate("bar"),

                  "Value not returned by Private.GetPrivate()")

    self.assertIs("egg", pDict.Unprivate()["bar"],

                  "Value not returned by PrivateDict.Unprivate()")

    self.assertIn(

      "egg",

      serializer.Dump(pDict,

                      private_encoder=serializer.EncodeWithPrivateFields)

    )

  def testDictGet(self):

    self.assertIs("tar", serializer.PrivateDict().GetPrivate("bar", "tar"),

                  "Private.GetPrivate() did not handle the default case")

  def testZeronessPrivate(self):

    self.assertTrue(serializer.Private("foo"),

                    "Private of non-empty string is false")

    self.assertFalse(serializer.Private(""), "Private empty string is true")

class TestCheckDoctests(unittest.TestCase):

  def testCheckSerializer(self):

    results = doctest.testmod(serializer)

    self.assertEquals(results.failed, 0, "Doctest failures detected")