Enforces strict manifest content.

tools/isolate/run_test_from_archive.py

 #!/usr/bin/env python
 # Copyright (c) 2012 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 """Reads a manifest, creates a tree of hardlinks and runs the test.
 
 Keeps a local cache.
 """
 
 import ctypes
 import json
 import logging
 import optparse
 import os
 import re
 import shutil
 import stat
 import subprocess
 import sys
 import tempfile
 import time
 import urllib
 
 
 # Types of action accepted by recreate_tree().
 HARDLINK, SYMLINK, COPY = range(1, 4)
 
+RE_IS_SHA1 = re.compile(r'^[a-fA-F0-9]{40}$')
+
+
+class ConfigError(ValueError):
+  """Generic failure to load a manifest."""
+  pass
+
 
 class MappingError(OSError):
   """Failed to recreate the tree."""
   pass
 
 
 def os_link(source, link_name):
   """Add support for os.link() on Windows."""
   if sys.platform == 'win32':
     if not ctypes.windll.kernel32.CreateHardLinkW(
@@ skipping 126 matching lines @@
 
 
 def make_temp_dir(prefix, root_dir):
   """Returns a temporary directory on the same file system than root_dir."""
   base_temp_dir = None
   if not is_same_filesystem(root_dir, tempfile.gettempdir()):
     base_temp_dir = os.path.dirname(root_dir)
   return tempfile.mkdtemp(prefix=prefix, dir=base_temp_dir)
 
 
+def assert_is_str(value):
+  if not isinstance(value, basestring):
+    raise ConfigError('Expected string, got %r' % value)
+
+
+def assert_is_int(value):
+  if not isinstance(value, int):
+    raise ConfigError('Expected int, got %r' % value)
+
+
+def assert_is_bool(value):
+  if value not in (None, True, False):
+    raise ConfigError('Expected bool, got %r' % value)
+
+
+def assert_is_sha1(value):
+  assert_is_str(value)
+  if not RE_IS_SHA1.match(value):
+    raise ConfigError('Expected sha-1, got %r' % value)
+
+
+def get_assert_is_list(check):
+  """Returns a function that asserts the value is a list."""
+  def is_list_item(value):
+    if not isinstance(value, list):
+      raise ConfigError('Expected list, got %r' % value)
+    any(check(item) for item in value)
+  return is_list_item
+
+
+def get_assert_is_dict(key_check, values_check):
+  """Returns a function that ensure the value is a dict.
+
+  It checks all the keys passes key_check and values_check is a dict of each
+  checks for each key.
+
+  If values_check contains a None entry, keys not in values_check will be
+  checked with this key. If None is not present in values_check, unknown keys
+  will be refused.
+  """
+  def is_dict_item(value):
+    if not isinstance(value, dict):
+      raise ConfigError('Expected dict, got %r' % value)
+
+    for key in value:
+      key_check(key)
+
+    # values_check is the set of valid keys.
+    unknown = set(value) - set(values_check)
+    if unknown and not None in values_check:
+      raise ConfigError(
+          'Got unknown key(s) %s in %r; expected only %s' %
+            (', '.join(unknown), value, ', '.join(values_check)))
+    for k, v in value.iteritems():
+      if k in values_check:
+        values_check[k](v)
+      else:
+        values_check[None](v)
+  return is_dict_item
+
+
+def load_manifest(content):
+  """Verifies the manifest is valid and loads this object with the json data.
+  """
+  data = json.loads(content)
+
+  file_keys = {
+    'link': assert_is_str,
+    'mode': assert_is_int,
+    'sha-1': assert_is_sha1,
+    'timestamp': assert_is_int,
+  }
+
+  files_key = {
+    # Any file has to fit files_keys specification.
+    None: get_assert_is_dict(assert_is_str, file_keys),
+  }
+
+  keys = {
+    'command': get_assert_is_list(assert_is_str),
+    # Could use assert_is_valid_filename instead of assert_is_str.
+    'files': get_assert_is_dict(assert_is_str, files_key),
+    'read_only': assert_is_bool,
+    'relative_cwd': assert_is_str,
+  }
+
+  get_assert_is_dict(assert_is_str, keys)(data)

[cmp, 2012/08/28 17:38:29]

this code requires a lot of state to be in-mind to verify it's correctness. 
this is possibly ironic, because it is designed to verify the correctness of its
input.  i would prefer that load_manifest be simpler / more straightforward. 
would that make it less fun to write/harder to maintain?

+  # Add a special verification.
+  for value in data.get('files', {}).itervalues():
+    if bool('sha-1' in value) and bool('link' in value):
+      raise ConfigError(
+          'Did not expect both \'sha-1\' and \'link\', got: %r' % value)
+  return data
+
+
 def fix_python_path(cmd):
   """Returns the fixed command line to call the right python executable."""
   out = cmd[:]
   if out[0] == 'python':
     out[0] = sys.executable
   elif out[0].endswith('.py'):
     out.insert(0, sys.executable)
   return out
 
 
@@ skipping 201 matching lines @@
             os.makedirs(outfiledir)
           if 'sha-1' in properties:
             # A normal file.
             infile = properties['sha-1']
             cache.retrieve(infile)
             link_file(outfile, cache.path(infile), HARDLINK)
           elif 'link' in properties:
             # A symlink.
             os.symlink(properties['link'], outfile)
           else:
-            raise ValueError('Unexpected entry: %s' % properties)
+            raise ConfigError('Unexpected entry: %s' % properties)
           if 'mode' in properties:
             # It's not set on Windows.
             os.chmod(outfile, properties['mode'])
 
       cwd = os.path.join(outdir, manifest.get('relative_cwd', ''))
       if not os.path.isdir(cwd):
         os.makedirs(cwd)
       if manifest.get('read_only'):
         make_writable(outdir, True)
       cmd = manifest['command']
@@ skipping 68 matching lines @@
     parser.error('One and only one of --manifest or --hash is required.')
   if not options.remote:
     parser.error('--remote is required.')
   if args:
     parser.error('Unsupported args %s' % ' '.join(args))
 
   if options.hash:
     # First calculate the reference to it.
     options.manifest = '%s/%s' % (options.remote.rstrip('/'), options.hash)
   try:
-    manifest = json.load(open_remote(options.manifest))
+    manifest = load_manifest(open_remote(options.manifest).read())
   except IOError as e:
     parser.error(
         'Failed to read manifest %s; remote:%s; hash:%s; %s' %
         (options.manifest, options.remote, options.hash, str(e)))
 
   policies = CachePolicies(
       options.max_cache_size, options.min_free_space, options.max_items)
   try:
     return run_tha_test(
         manifest,
         os.path.abspath(options.cache),
         options.remote,
         policies)
-  except MappingError, e:
+  except (ConfigError, MappingError), e:
     print >> sys.stderr, str(e)
     return 1
 
 
 if __name__ == '__main__':
   sys.exit(main())