Docserver: Fix invalid path usage in CloudStorageFileSystem.

chrome/common/extensions/docs/server2/gcs_file_system.py

 # Copyright 2014 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.
 
 from third_party.cloudstorage import cloudstorage_api
 from third_party.cloudstorage import common
 from third_party.cloudstorage import errors
 
 from docs_server_utils import StringIdentity
 from file_system import FileSystem, FileNotFoundError, StatInfo
 from future import Gettable, Future
+from path_util import (
+    AssertIsDirectory, AssertIsFile, AssertIsValid, IsDirectory, Join)
 
 import logging
 import traceback
 
+
+# See gcs_file_system_provider.py for documentation on using Google Cloud
+# Storage as a filesystem.
+#
+# Note that the path requirements for GCS are different for the docserver;
+# GCS requires that paths start with a /, we require that they don't.
+
+
 # Name of the file containing the Git hash of the latest commit sync'ed
 # to Cloud Storage. This file is generated by the Github->GCS sync script
-LAST_COMMIT_HASH_FILENAME='.__lastcommit.txt'
+LAST_COMMIT_HASH_FILENAME = '.__lastcommit.txt'
 
-'''See gcs_file_system_provider.py for documentation on using Google Cloud
-Storage as a filesystem.
-'''
 def _ReadFile(filename):
+  AssertIsFile(filename)
   try:
-    with cloudstorage_api.open(filename, 'r') as f:
+    with cloudstorage_api.open('/' + filename, 'r') as f:
       return f.read()
   except errors.Error:
     raise FileNotFoundError('Read failed for %s: %s' % (filename,
         traceback.format_exc()))
 
 def _ListDir(dir_name):
+  AssertIsDirectory(dir_name)
   try:
-    files = cloudstorage_api.listbucket(dir_name)
-    return [os_path.filename for os_path in files]
+    files = cloudstorage_api.listbucket('/' + dir_name)
+    return [os_path.filename.lstrip('/') for os_path in files]
   except errors.Error:
     raise FileNotFoundError('cloudstorage.listbucket failed for %s: %s' %
                             (dir_name, traceback.format_exc()))
 
 def _CreateStatInfo(bucket, path):
-  bucket = '/%s' % bucket
-  full_path = '/'.join( (bucket, path.lstrip('/')) )
-  last_commit_file = '%s/%s' % (bucket, LAST_COMMIT_HASH_FILENAME)
+  full_path = Join(bucket, path)
+  last_commit_file = Join(bucket, LAST_COMMIT_HASH_FILENAME)
   try:
     last_commit = _ReadFile(last_commit_file)
-    if full_path.endswith('/'):
+    if IsDirectory(full_path):
       child_versions = dict()
       # Fetching stats for all files under full_path, recursively. The
       # listbucket method uses a prefix approach to simulate hierarchy,
       # but calling it without the "delimiter" argument searches for prefix,
       # which means, for directories, everything beneath it.
-      for _file in cloudstorage_api.listbucket(full_path):
-        filename = _file.filename[len(full_path):]
+      for _file in cloudstorage_api.listbucket('/' + full_path):
+        filename = _file.filename.lstrip('/')[len(full_path):]
         child_versions[filename] = last_commit
     else:
       child_versions = None
     return StatInfo(last_commit, child_versions)
   except (TypeError, errors.Error):
     raise FileNotFoundError('cloudstorage.stat failed for %s: %s' % (path,
                             traceback.format_exc()))
 
 class CloudStorageFileSystem(FileSystem):
   '''FileSystem implementation which fetches resources from Google Cloud
   Storage.
   '''
   def __init__(self, bucket, debug_access_token=None, debug_bucket_prefix=None):
     self._bucket = bucket
     if debug_access_token:
       logging.debug('gcs: using debug access token: %s' % debug_access_token)
       common.set_access_token(debug_access_token)
     if debug_bucket_prefix:
       logging.debug('gcs: prefixing all bucket names with %s' %
                     debug_bucket_prefix)
       self._bucket = debug_bucket_prefix + self._bucket
+    AssertIsValid(self._bucket)
 
   def Read(self, paths):
     def resolve():
       try:
         result = {}
         for path in paths:
-          full_path = '/%s/%s' % (self._bucket, path.lstrip('/'))
-          logging.debug('gcs: requested path %s, reading %s' %
+          full_path = Join(self._bucket, path)
+          logging.debug('gcs: requested path "%s", reading "%s"' %
                         (path, full_path))
-          if path == '' or path.endswith('/'):
+          if IsDirectory(path):
             result[path] = _ListDir(full_path)
           else:
             result[path] = _ReadFile(full_path)
         return result
       except errors.AuthorizationError:
         self._warnAboutAuthError()
         raise
 
     return Future(delegate=Gettable(resolve))
 
   def Refresh(self):
     return Future(value=())
 
   def Stat(self, path):
+    AssertIsValid(path)
     try:
       return _CreateStatInfo(self._bucket, path)
     except errors.AuthorizationError:
       self._warnAboutAuthError()
       raise
 
   def GetIdentity(self):
     return '@'.join((self.__class__.__name__, StringIdentity(self._bucket)))
 
   def __repr__(self):
     return 'LocalFileSystem(%s)' % self._bucket
 
   def _warnAboutAuthError(self):
     logging.warn(('Authentication error on Cloud Storage. Check if your'
                   ' appengine project has permissions to Read the GCS'
                   ' buckets. If you are running a local appengine server,'
                   ' you need to set an access_token in'
                   ' local_debug/gcs_debug.conf.'
                   ' Remember that this token expires in less than 10'
                   ' minutes, so keep it updated. See'
                   ' gcs_file_system_provider.py for instructions.'));
     logging.debug(traceback.format_exc())