¡@
OpenStack Study: constraints.py**** CubicPower OpenStack Study **** def reload_constraints(): """ Parse SWIFT_CONF_FILE and reset module level global contraint attrs, populating OVERRIDE_CONSTRAINTS AND EFFECTIVE_CONSTRAINTS along the way. """ global SWIFT_CONSTRAINTS_LOADED, OVERRIDE_CONSTRAINTS SWIFT_CONSTRAINTS_LOADED = False OVERRIDE_CONSTRAINTS = {} constraints_conf = ConfigParser() if constraints_conf.read(SWIFT_CONF_FILE): SWIFT_CONSTRAINTS_LOADED = True for name in DEFAULT_CONSTRAINTS: try: value = int(constraints_conf.get('swift-constraints', name)) except (NoSectionError, NoOptionError): pass else: OVERRIDE_CONSTRAINTS[name] = value for name, default in DEFAULT_CONSTRAINTS.items(): value = OVERRIDE_CONSTRAINTS.get(name, default) EFFECTIVE_CONSTRAINTS[name] = value # "globals" in this context is module level globals, always. globals()[name.upper()] = value reload_constraints() # Maximum slo segments in buffer MAX_BUFFERED_SLO_SEGMENTS = 10000 #: Query string format= values to their corresponding content-type values FORMAT2CONTENT_TYPE = {'plain': 'text/plain', 'json': 'application/json', 'xml': 'application/xml'}
**** CubicPower OpenStack Study **** def check_metadata(req, target_type): """ Check metadata sent in the request headers. :param req: request object :param target_type: str: one of: object, container, or account: indicates which type the target storage for the metadata is :returns: HTTPBadRequest with bad metadata otherwise None """ prefix = 'x-%s-meta-' % target_type.lower() meta_count = 0 meta_size = 0 for key, value in req.headers.iteritems(): if isinstance(value, basestring) and len(value) > MAX_HEADER_SIZE: return HTTPBadRequest(body='Header value too long: %s' % key[:MAX_META_NAME_LENGTH], request=req, content_type='text/plain') if not key.lower().startswith(prefix): continue key = key[len(prefix):] if not key: return HTTPBadRequest(body='Metadata name cannot be empty', request=req, content_type='text/plain') meta_count += 1 meta_size += len(key) + len(value) if len(key) > MAX_META_NAME_LENGTH: return HTTPBadRequest( body='Metadata name too long: %s%s' % (prefix, key), request=req, content_type='text/plain') elif len(value) > MAX_META_VALUE_LENGTH: return HTTPBadRequest( body='Metadata value longer than %d: %s%s' % ( MAX_META_VALUE_LENGTH, prefix, key), request=req, content_type='text/plain') elif meta_count > MAX_META_COUNT: return HTTPBadRequest( body='Too many metadata items; max %d' % MAX_META_COUNT, request=req, content_type='text/plain') elif meta_size > MAX_META_OVERALL_SIZE: return HTTPBadRequest( body='Total metadata too large; max %d' % MAX_META_OVERALL_SIZE, request=req, content_type='text/plain') return None
**** CubicPower OpenStack Study **** def check_object_creation(req, object_name): """ Check to ensure that everything is alright about an object to be created. :param req: HTTP request object :param object_name: name of object to be created :returns HTTPRequestEntityTooLarge: the object is too large :returns HTTPLengthRequired: missing content-length header and not a chunked request :returns HTTPBadRequest: missing or bad content-type header, or bad metadata """ if req.content_length and req.content_length > MAX_FILE_SIZE: return HTTPRequestEntityTooLarge(body='Your request is too large.', request=req, content_type='text/plain') if req.content_length is None and \ req.headers.get('transfer-encoding') != 'chunked': return HTTPLengthRequired(request=req) if 'X-Copy-From' in req.headers and req.content_length: return HTTPBadRequest(body='Copy requests require a zero byte body', request=req, content_type='text/plain') if len(object_name) > MAX_OBJECT_NAME_LENGTH: return HTTPBadRequest(body='Object name length of %d longer than %d' % (len(object_name), MAX_OBJECT_NAME_LENGTH), request=req, content_type='text/plain') if 'Content-Type' not in req.headers: return HTTPBadRequest(request=req, content_type='text/plain', body='No content type') if not check_utf8(req.headers['Content-Type']): return HTTPBadRequest(request=req, body='Invalid Content-Type', content_type='text/plain') return check_metadata(req, 'object')
**** CubicPower OpenStack Study **** def check_mount(root, drive): """ Verify that the path to the device is a mount point and mounted. This allows us to fast fail on drives that have been unmounted because of issues, and also prevents us for accidentally filling up the root partition. :param root: base path where the devices are mounted :param drive: drive name to be checked :returns: True if it is a valid mounted device, False otherwise """ if not (urllib.quote_plus(drive) == drive): return False path = os.path.join(root, drive) return ismount(path)
**** CubicPower OpenStack Study **** def check_float(string): """ Helper function for checking if a string can be converted to a float. :param string: string to be verified as a float :returns: True if the string can be converted to a float, False otherwise """ try: float(string) return True except ValueError: return False
**** CubicPower OpenStack Study **** def check_utf8(string): """ Validate if a string is valid UTF-8 str or unicode and that it does not contain any null character. :param string: string to be validated :returns: True if the string is valid utf-8 str or unicode and contains no null characters, False otherwise """ if not string: return False try: if isinstance(string, unicode): string.encode('utf-8') else: string.decode('UTF-8') return '\x00' not in string # If string is unicode, decode() will raise UnicodeEncodeError # So, we should catch both UnicodeDecodeError & UnicodeEncodeError except UnicodeError: return False
**** CubicPower OpenStack Study **** def check_copy_from_header(req): """ Validate that the value from x-copy-from header is well formatted. We assume the caller ensures that x-copy-from header is present in req.headers. :param req: HTTP request object :returns: A tuple with container name and object name :raise: HTTPPreconditionFailed if x-copy-from value is not well formatted. """ src_header = unquote(req.headers.get('X-Copy-From')) if not src_header.startswith('/'): src_header = '/' + src_header try: return split_path(src_header, 2, 2, True) except ValueError: raise HTTPPreconditionFailed( request=req, body='X-Copy-From header must be of the form' '
|