python/google/appengine/ext/cloudstorage/stub_dispatcher.py

   1 #!/usr/bin/env python
   2 #
   3 # Copyright 2007 Google Inc.
   4 #
   5 # Licensed under the Apache License, Version 2.0 (the "License");
   6 # you may not use this file except in compliance with the License.
   7 # You may obtain a copy of the License at
   8 #
   9 #     http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 # Unless required by applicable law or agreed to in writing, software
  12 # distributed under the License is distributed on an "AS IS" BASIS,
  13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16 #
  17 """Dispatcher to handle Google Cloud Storage stub requests."""
  18
  19
  20
  21
  22
  23
  24
  25
  26
  27 import httplib
  28 import re
  29 import urllib
  30 import urlparse
  31 import xml.etree.ElementTree as ET
  32
  33 from google.appengine.api import apiproxy_stub_map
  34 from google.appengine.api import urlfetch_stub
  35 from google.appengine.ext.cloudstorage import cloudstorage_stub
  36 from google.appengine.ext.cloudstorage import common
  37
  38
  39
  40 _MAX_GET_BUCKET_RESULT = 1000
  41
  42
  43
  44 def _urlfetch_to_gcs_stub(url, payload, method, headers, request, response,
  45                           follow_redirects=False, deadline=None,
  46                           validate_certificate=None):
  47
  48   """Forwards gcs urlfetch requests to gcs_dispatcher.
  49
  50   See apphosting.api.urlfetch_stub.URLFetchServiceStub._RetrieveURL.
  51   """
  52   headers_map = dict(
  53       (header.key().lower(), header.value()) for header in headers)
  54   result = dispatch(method, headers_map, url, payload)
  55   response.set_statuscode(result.status_code)
  56   response.set_content(result.content[:urlfetch_stub.MAX_RESPONSE_SIZE])
  57   for k, v in result.headers.iteritems():
  58     if k.lower() == 'content-length' and method != 'HEAD':
  59       v = len(response.content())
  60     header_proto = response.add_header()
  61     header_proto.set_key(k)
  62     header_proto.set_value(str(v))
  63   if len(result.content) > urlfetch_stub.MAX_RESPONSE_SIZE:
  64     response.set_contentwastruncated(True)
  65
  66
  67 def _urlmatcher_for_gcs_stub(url):
  68   """Determines whether a url should be handled by gcs stub."""
  69   _, host, _, _, _ = urlparse.urlsplit(url)
  70   return host == common.LOCAL_API_HOST
  71
  72
  73
  74 URLMATCHERS_TO_FETCH_FUNCTIONS = [
  75     (_urlmatcher_for_gcs_stub, _urlfetch_to_gcs_stub)]
  76
  77
  78 class _FakeUrlFetchResult(object):
  79   def __init__(self, status, headers, content):
  80     self.status_code = status
  81     self.headers = headers
  82     self.content = content
  83
  84
  85 def dispatch(method, headers, url, payload):
  86   """Dispatches incoming request and returns response.
  87
  88   In dev appserver or unittest environment, this method is called instead of
  89   urlfetch.
  90
  91   Args:
  92     method: urlfetch method.
  93     headers: urlfetch headers.
  94     url: urlfetch url.
  95     payload: urlfecth payload.
  96
  97   Returns:
  98     A _FakeUrlFetchResult.
  99
 100   Raises:
 101     ValueError: invalid request method.
 102   """
 103   method, headers, filename, param_dict = _preprocess(method, headers, url)
 104   gcs_stub = cloudstorage_stub.CloudStorageStub(
 105       apiproxy_stub_map.apiproxy.GetStub('blobstore').storage)
 106
 107   if method == 'POST':
 108     return _handle_post(gcs_stub, filename, headers)
 109   elif method == 'PUT':
 110     return _handle_put(gcs_stub, filename, param_dict, headers, payload)
 111   elif method == 'GET':
 112     return _handle_get(gcs_stub, filename, param_dict, headers)
 113   elif method == 'HEAD':
 114     return _handle_head(gcs_stub, filename)
 115   elif method == 'DELETE':
 116     return _handle_delete(gcs_stub, filename)
 117   raise ValueError('Unrecognized request method %r.' % method)
 118
 119
 120 def _preprocess(method, headers, url):
 121   """Unify input.
 122
 123   Example:
 124     _preprocess('POST', {'Content-Type': 'Foo'}, http://gcs.com/b/f?foo=bar)
 125     -> 'POST', {'content-type': 'Foo'}, '/b/f', {'foo':'bar'}
 126
 127   Args:
 128     method: HTTP method used by the request.
 129     headers: HTTP request headers in a dict.
 130     url: HTTP request url.
 131
 132   Returns:
 133     method: method in all upper case.
 134     headers: headers with keys in all lower case.
 135     filename: a google storage filename of form /bucket/filename or
 136       a bucket path of form /bucket
 137     param_dict: a dict of query parameters.
 138   """
 139   _, _, filename, query, _ = urlparse.urlsplit(url)
 140   param_dict = urlparse.parse_qs(query, True)
 141   for k in param_dict:
 142     param_dict[k] = urllib.unquote(param_dict[k][0])
 143
 144   headers = dict((k.lower(), v) for k, v in headers.iteritems())
 145   return method, headers, filename, param_dict
 146
 147
 148 def _handle_post(gcs_stub, filename, headers):
 149   """Handle POST that starts object creation."""
 150   content_type = _ContentType(headers)
 151   token = gcs_stub.post_start_creation(filename, headers)
 152   response_headers = {
 153       'location': 'https://storage.googleapis.com/%s?%s' % (
 154           filename,
 155           urllib.urlencode({'upload_id': token})),
 156       'content-type': content_type.value,
 157       'content-length': 0
 158   }
 159   return _FakeUrlFetchResult(201, response_headers, '')
 160
 161
 162 def _handle_put(gcs_stub, filename, param_dict, headers, payload):
 163   """Handle PUT that continues object creation."""
 164   token = _get_param('upload_id', param_dict)
 165   content_range = _ContentRange(headers)
 166
 167   if not content_range.value:
 168     raise ValueError('Missing header content-range.')
 169
 170   gcs_stub.put_continue_creation(token,
 171                                 payload,
 172                                 content_range.range,
 173                                 content_range.last)
 174   if content_range.last:
 175     filestat = gcs_stub.head_object(filename)
 176     response_headers = {
 177         'content-length': filestat.st_size,
 178     }
 179     response_status = httplib.OK
 180   else:
 181     response_headers = {}
 182     response_status = 308
 183
 184   return _FakeUrlFetchResult(response_status, response_headers, '')
 185
 186
 187 def _handle_get(gcs_stub, filename, param_dict, headers):
 188   """Handle GET object and GET bucket."""
 189   if filename.rfind('/') == 0:
 190
 191     return _handle_get_bucket(gcs_stub, filename, param_dict)
 192   else:
 193
 194     result = _handle_head(gcs_stub, filename)
 195     if result.status_code == 404:
 196       return result
 197     start, end = _Range(headers).value
 198     st_size = result.headers['content-length']
 199     if end is None:
 200       end = st_size - 1
 201     result.headers['content-range'] = 'bytes: %d-%d/%d' % (start,
 202                                                            end,
 203                                                            st_size)
 204     result.content = gcs_stub.get_object(filename, start, end)
 205     return result
 206
 207
 208 def _handle_get_bucket(gcs_stub, bucketpath, param_dict):
 209   """Handle get bucket request."""
 210   prefix = _get_param('prefix', param_dict, '')
 211   max_keys = _get_param('max-keys', param_dict, _MAX_GET_BUCKET_RESULT)
 212   marker = _get_param('marker', param_dict, '')
 213
 214   stats = gcs_stub.get_bucket(bucketpath,
 215                               prefix,
 216                               marker,
 217                               max_keys)
 218
 219   builder = ET.TreeBuilder()
 220   builder.start('ListBucketResult', {'xmlns': common.CS_XML_NS})
 221   last_object_name = ''
 222   for stat in stats:
 223     builder.start('Contents', {})
 224
 225     builder.start('Key', {})
 226     last_object_name = stat.filename[len(bucketpath) + 1:]
 227     builder.data(last_object_name)
 228     builder.end('Key')
 229
 230     builder.start('LastModified', {})
 231     builder.data(common.posix_to_dt_str(stat.st_ctime))
 232     builder.end('LastModified')
 233
 234     builder.start('ETag', {})
 235     builder.data(stat.etag)
 236     builder.end('ETag')
 237
 238     builder.start('Size', {})
 239     builder.data(str(stat.st_size))
 240     builder.end('Size')
 241
 242     builder.end('Contents')
 243
 244   if last_object_name:
 245     builder.start('NextMarker', {})
 246     builder.data(last_object_name)
 247     builder.end('NextMarker')
 248
 249   max_keys = _get_param('max-keys', param_dict)
 250   if max_keys is not None:
 251     builder.start('MaxKeys', {})
 252     builder.data(str(max_keys))
 253     builder.end('MaxKeys')
 254
 255   builder.end('ListBucketResult')
 256   root = builder.close()
 257
 258   body = ET.tostring(root)
 259   response_headers = {'content-length': len(body),
 260                       'content-type': 'application/xml'}
 261   return _FakeUrlFetchResult(200, response_headers, body)
 262
 263
 264 def _handle_head(gcs_stub, filename):
 265   """Handle HEAD request."""
 266   filestat = gcs_stub.head_object(filename)
 267   if not filestat:
 268     return _FakeUrlFetchResult(404, {}, '')
 269
 270   http_time = common.posix_time_to_http(filestat.st_ctime)
 271
 272   response_headers = {
 273       'content-length': filestat.st_size,
 274       'content-type': filestat.content_type,
 275       'etag': filestat.etag,
 276       'last-modified': http_time
 277   }
 278
 279   if filestat.metadata:
 280     response_headers.update(filestat.metadata)
 281
 282   return _FakeUrlFetchResult(200, response_headers, '')
 283
 284
 285 def _handle_delete(gcs_stub, filename):
 286   """Handle DELETE object."""
 287   if gcs_stub.delete_object(filename):
 288     return _FakeUrlFetchResult(204, {}, '')
 289   else:
 290     return _FakeUrlFetchResult(404, {}, '')
 291
 292
 293 class _Header(object):
 294   """Wrapper class for a header.
 295
 296   A subclass helps to parse a specific header.
 297   """
 298
 299   HEADER = ''
 300   DEFAULT = None
 301
 302   def __init__(self, headers):
 303     """Initialize.
 304
 305     Initializes self.value to the value in request header, or DEFAULT if
 306     not defined in headers.
 307
 308     Args:
 309       headers: request headers.
 310     """
 311     self.value = self.DEFAULT
 312     for k in headers:
 313       if k.lower() == self.HEADER.lower():
 314         self.value = headers[k]
 315         break
 316
 317
 318 class _ContentType(_Header):
 319   """Content-type header."""
 320
 321   HEADER = 'Content-Type'
 322   DEFAULT = 'binary/octet-stream'
 323
 324
 325 class _ContentRange(_Header):
 326   """Content-Range header.
 327
 328   Used by resumable upload of unknown size. Possible formats:
 329     Content-Range: bytes 1-3/* (for uploading of unknown size)
 330     Content-Range: bytes */5 (for finalizing with no data)
 331   """
 332
 333   HEADER = 'Content-Range'
 334   RE_PATTERN = re.compile(r'^bytes (([0-9]+)-([0-9]+)|\*)/([0-9]+|\*)$')
 335
 336   def __init__(self, headers):
 337     super(_ContentRange, self).__init__(headers)
 338     if self.value:
 339       result = self.RE_PATTERN.match(self.value)
 340       if not result:
 341         raise ValueError('Invalid content-range header %s' % self.value)
 342
 343       self.no_data = result.group(1) == '*'
 344       self.last = result.group(4) != '*'
 345       if self.no_data and not self.last:
 346         raise ValueError('Invalid content-range header %s' % self.value)
 347
 348       self.range = None
 349       if not self.no_data:
 350         self.range = (long(result.group(2)), long(result.group(3)))
 351
 352
 353 class _Range(_Header):
 354   """_Range header.
 355
 356   Used by read. Format: Range: bytes=1-3.
 357   """
 358
 359   HEADER = 'Range'
 360
 361   def __init__(self, headers):
 362     super(_Range, self).__init__(headers)
 363     if self.value:
 364       start, end = self.value.rsplit('=', 1)[-1].split('-')
 365       start, end = long(start), long(end)
 366     else:
 367       start, end = 0, None
 368     self.value = start, end
 369
 370
 371 def _get_param(param, param_dict, default=None):
 372   """Gets a parameter value from request query parameters.
 373
 374   Args:
 375     param: name of the parameter to get.
 376     param_dict: a dict of request query parameters.
 377     default: default value if not defined.
 378
 379   Returns:
 380     Value of the parameter or default if not defined.
 381   """
 382   result = param_dict.get(param, default)
 383   if param in ['max-keys'] and result:
 384     return long(result)
 385   return result