def enforce_byte_count(inner_iter, nbytes):
    """
    Enforces that inner_iter yields exactly <nbytes> bytes before
    exhaustion.

    If inner_iter fails to do so, BadResponseLength is raised.

    :param inner_iter: iterable of bytestrings
    :param nbytes: number of bytes expected
    """
    try:
        bytes_left = nbytes
        for chunk in inner_iter:
            if bytes_left >= len(chunk):
                yield chunk
                bytes_left -= len(chunk)
            else:
                yield chunk[:bytes_left]
                raise BadResponseLength(
                    "Too many bytes; truncating after %d bytes "
                    "with at least %d surplus bytes remaining" % (
                        nbytes, len(chunk) - bytes_left))

        if bytes_left:
            raise BadResponseLength('Expected another %d bytes' % (
                bytes_left,))
    finally:
        close_if_possible(inner_iter)

    def _fetch_sub_slo_segments(self, req, version, acc, con, obj):
        """
        Fetch the submanifest, parse it, and return it.
        Raise exception on failures.
        """
        sub_req = make_subrequest(
            req.environ,
            path="/".join(["", version, acc, con, obj]),
            method="GET",
            headers={"x-auth-token": req.headers.get("x-auth-token")},
            agent=("%(orig)s " + "SLO MultipartGET"),
            swift_source="SLO",
        )
        sub_resp = sub_req.get_response(self.slo.app)

        if not is_success(sub_resp.status_int):
            close_if_possible(sub_resp.app_iter)
            raise ListingIterError(
                "ERROR: while fetching %s, GET of submanifest %s "
                "failed with status %d" % (req.path, sub_req.path, sub_resp.status_int)
            )

        try:
            with closing_if_possible(sub_resp.app_iter):
                return json.loads("".join(sub_resp.app_iter))
        except ValueError as err:
            raise ListingIterError(
                "ERROR: while fetching %s, JSON-decoding of submanifest %s "
                "failed with %s" % (req.path, sub_req.path, err)
            )

    def _get_source_object(self, ssc_ctx, source_path, req):
        source_req = req.copy_get()

        # make sure the source request uses it's container_info
        source_req.headers.pop('X-Backend-Storage-Policy-Index', None)
        source_req.path_info = quote(source_path)
        source_req.headers['X-Newest'] = 'true'
        if 'swift.post_as_copy' in req.environ:
            # We're COPYing one object over itself because of a POST; rely on
            # the PUT for write authorization, don't require read authorization
            source_req.environ['swift.authorize'] = lambda req: None
            source_req.environ['swift.authorize_override'] = True

        # in case we are copying an SLO manifest, set format=raw parameter
        params = source_req.params
        if params.get('multipart-manifest') == 'get':
            params['format'] = 'raw'
            source_req.params = params

        source_resp = ssc_ctx.get_source_resp(source_req)

        if source_resp.content_length is None:
            # This indicates a transfer-encoding: chunked source object,
            # which currently only happens because there are more than
            # CONTAINER_LISTING_LIMIT segments in a segmented object. In
            # this case, we're going to refuse to do the server-side copy.
            close_if_possible(source_resp.app_iter)
            return HTTPRequestEntityTooLarge(request=req)

        if source_resp.content_length > MAX_FILE_SIZE:
            close_if_possible(source_resp.app_iter)
            return HTTPRequestEntityTooLarge(request=req)

        return source_resp

 def object_request(self, req, api_version, account, container, obj,
                    allow_versioned_writes):
     container_name = unquote(container)
     object_name = unquote(obj)
     orig_container = get_unversioned_container(container_name)
     if orig_container != container_name:
         orig_object, version = \
             swift3_split_object_name_version(object_name)
         req.environ['oio.query'] = {'version': version}
         req.environ['PATH_INFO'] = '/%s/%s/%s/%s' % (api_version,
                                                      account,
                                                      quote(orig_container),
                                                      quote(orig_object))
     elif req.method == 'DELETE':
         ver_mode = req.headers.get('X-Backend-Versioning-Mode-Override',
                                    'history')
         if ver_mode == 'stack':
             # Do not create a delete marker, delete the latest version
             obj_inf = get_object_info(req.environ, self.app,
                                       swift_source='VW')
             req.environ['oio.query'] = {
                 'version': obj_inf.get('sysmeta', {}).get('version-id')
             }
     resp = req.get_response(self.app)
     if req.method == 'HEAD':
         close_if_possible(resp.app_iter)
     return resp

    def _get_source_object(self, ssc_ctx, source_path, req):
        source_req = req.copy_get()

        # make sure the source request uses it's container_info
        source_req.headers.pop('X-Backend-Storage-Policy-Index', None)
        source_req.path_info = source_path
        source_req.headers['X-Newest'] = 'true'

        # in case we are copying an SLO manifest, set format=raw parameter
        params = source_req.params
        if params.get('multipart-manifest') == 'get':
            params['format'] = 'raw'
            source_req.params = params

        source_resp = ssc_ctx.get_source_resp(source_req)

        if source_resp.content_length is None:
            # This indicates a transfer-encoding: chunked source object,
            # which currently only happens because there are more than
            # CONTAINER_LISTING_LIMIT segments in a segmented object. In
            # this case, we're going to refuse to do the server-side copy.
            close_if_possible(source_resp.app_iter)
            return HTTPRequestEntityTooLarge(request=req)

        if source_resp.content_length > MAX_FILE_SIZE:
            close_if_possible(source_resp.app_iter)
            return HTTPRequestEntityTooLarge(request=req)

        return source_resp

    def _fetch_sub_slo_segments(self, req, version, acc, con, obj):
        """
        Fetch the submanifest, parse it, and return it.
        Raise exception on failures.
        """
        sub_req = make_subrequest(
            req.environ, path='/'.join(['', version, acc, con, obj]),
            method='GET',
            headers={'x-auth-token': req.headers.get('x-auth-token')},
            agent=('%(orig)s ' + 'SLO MultipartGET'), swift_source='SLO')
        sub_resp = sub_req.get_response(self.slo.app)

        if not is_success(sub_resp.status_int):
            close_if_possible(sub_resp.app_iter)
            raise ListingIterError(
                'ERROR: while fetching %s, GET of submanifest %s '
                'failed with status %d' % (req.path, sub_req.path,
                                           sub_resp.status_int))

        try:
            with closing_if_possible(sub_resp.app_iter):
                return json.loads(''.join(sub_resp.app_iter))
        except ValueError as err:
            raise ListingIterError(
                'ERROR: while fetching %s, JSON-decoding of submanifest %s '
                'failed with %s' % (req.path, sub_req.path, err))

 def close(self):
     """
     Called when the client disconnect. Ensure that the connection to the
     backend server is closed.
     """
     if self.current_resp:
         close_if_possible(self.current_resp.app_iter)

    def _listing_pages_iter(self, account_name, lcontainer, lprefix,
                            env, marker='', end_marker='', reverse=True):
        '''Get "pages" worth of objects that start with a prefix.

        The optional keyword arguments ``marker``, ``end_marker``, and
        ``reverse`` are used similar to how they are for containers. We're
        either coming:

           - directly from ``_listing_iter``, in which case none of the
             optional args are specified, or

           - from ``_in_proxy_reverse_listing``, in which case ``reverse``
             is ``False`` and both ``marker`` and ``end_marker`` are specified
             (although they may still be blank).
        '''
        while True:
            lreq = make_pre_authed_request(
                env, method='GET', swift_source='VW',
                path='/v1/%s/%s' % (account_name, lcontainer))
            lreq.environ['QUERY_STRING'] = \
                'format=json&prefix=%s&marker=%s' % (
                    quote(lprefix), quote(marker))
            if end_marker:
                lreq.environ['QUERY_STRING'] += '&end_marker=%s' % (
                    quote(end_marker))
            if reverse:
                lreq.environ['QUERY_STRING'] += '&reverse=on'
            lresp = lreq.get_response(self.app)
            if not is_success(lresp.status_int):
                close_if_possible(lresp.app_iter)
                if lresp.status_int == HTTP_NOT_FOUND:
                    raise ListingIterNotFound()
                elif is_client_error(lresp.status_int):
                    raise HTTPPreconditionFailed()
                else:
                    raise ListingIterError()

            if not lresp.body:
                break

            sublisting = json.loads(lresp.body)
            if not sublisting:
                break

            # When using the ``reverse`` param, check that the listing is
            # actually reversed
            first_item = sublisting[0]['name'].encode('utf-8')
            last_item = sublisting[-1]['name'].encode('utf-8')
            page_is_after_marker = marker and first_item > marker
            if reverse and (first_item < last_item or page_is_after_marker):
                # Apparently there's at least one pre-2.6.0 container server
                yield self._in_proxy_reverse_listing(
                    account_name, lcontainer, lprefix,
                    env, marker, sublisting)
                return

            marker = last_item
            yield sublisting

    def handle_slo_get_or_head(self, req, start_response):
        """
        Takes a request and a start_response callable and does the normal WSGI
        thing with them. Returns an iterator suitable for sending up the WSGI
        chain.

        :param req: swob.Request object; is a GET or HEAD request aimed at
                    what may be a static large object manifest (or may not).
        :param start_response: WSGI start_response callable
        """
        resp_iter = self._app_call(req.environ)

        # make sure this response is for a static large object manifest
        for header, value in self._response_headers:
            if (header.lower() == 'x-static-large-object' and
                    config_true_value(value)):
                break
        else:
            # Not a static large object manifest. Just pass it through.
            start_response(self._response_status,
                           self._response_headers,
                           self._response_exc_info)
            return resp_iter

        # Handle pass-through request for the manifest itself
        if req.params.get('multipart-manifest') == 'get':
            new_headers = []
            for header, value in self._response_headers:
                if header.lower() == 'content-type':
                    new_headers.append(('Content-Type',
                                        'application/json; charset=utf-8'))
                else:
                    new_headers.append((header, value))
            self._response_headers = new_headers
            start_response(self._response_status,
                           self._response_headers,
                           self._response_exc_info)
            return resp_iter

        if self._need_to_refetch_manifest(req):
            req.environ['swift.non_client_disconnect'] = True
            close_if_possible(resp_iter)
            del req.environ['swift.non_client_disconnect']

            get_req = make_subrequest(
                req.environ, method='GET',
                headers={'x-auth-token': req.headers.get('x-auth-token')},
                agent=('%(orig)s ' + 'SLO MultipartGET'), swift_source='SLO')
            resp_iter = self._app_call(get_req.environ)

        # Any Content-Range from a manifest is almost certainly wrong for the
        # full large object.
        resp_headers = [(h, v) for h, v in self._response_headers
                        if not h.lower() == 'content-range']

        response = self.get_or_head_response(
            req, resp_headers, resp_iter)
        return response(req.environ, start_response)

    def handle_slo_get_or_head(self, req, start_response):
        """
        Takes a request and a start_response callable and does the normal WSGI
        thing with them. Returns an iterator suitable for sending up the WSGI
        chain.

        :param req: swob.Request object; is a GET or HEAD request aimed at
                    what may be a static large object manifest (or may not).
        :param start_response: WSGI start_response callable
        """
        resp_iter = self._app_call(req.environ)

        # make sure this response is for a static large object manifest
        for header, value in self._response_headers:
            if (header.lower() == 'x-static-large-object' and
                    config_true_value(value)):
                break
        else:
            # Not a static large object manifest. Just pass it through.
            start_response(self._response_status,
                           self._response_headers,
                           self._response_exc_info)
            return resp_iter

        # Handle pass-through request for the manifest itself
        if req.params.get('multipart-manifest') == 'get':
            new_headers = []
            for header, value in self._response_headers:
                if header.lower() == 'content-type':
                    new_headers.append(('Content-Type',
                                        'application/json; charset=utf-8'))
                else:
                    new_headers.append((header, value))
            self._response_headers = new_headers
            start_response(self._response_status,
                           self._response_headers,
                           self._response_exc_info)
            return resp_iter

        # Just because a response shows that an object is a SLO manifest does
        # not mean that response's body contains the entire SLO manifest. If
        # it doesn't, we need to make a second request to actually get the
        # whole thing.
        if req.method == 'HEAD' or req.range:
            req.environ['swift.non_client_disconnect'] = True
            close_if_possible(resp_iter)
            del req.environ['swift.non_client_disconnect']

            get_req = req.copy_get()
            get_req.range = None
            get_req.environ['swift.source'] = 'SLO'
            get_req.user_agent = "%s SLO MultipartGET" % get_req.user_agent
            resp_iter = self._app_call(get_req.environ)

        response = self.get_or_head_response(req, self._response_headers,
                                             resp_iter)
        return response(req.environ, start_response)

 def _check_response_error(self, req, resp):
     """
     Raise Error Response in case of error
     """
     if is_success(resp.status_int):
         return
     close_if_possible(resp.app_iter)
     if is_client_error(resp.status_int):
         # missing container or bad permissions
         raise HTTPPreconditionFailed(request=req)
     # could not version the data, bail
     raise HTTPServiceUnavailable(request=req)

 def _put_versioned_obj(self, req, put_path_info, source_resp):
     # Create a new Request object to PUT to the versions container, copying
     # all headers from the source object apart from x-timestamp.
     put_req = make_pre_authed_request(
         req.environ, path=put_path_info, method='PUT',
         swift_source='VW')
     copy_header_subset(source_resp, put_req,
                        lambda k: k.lower() != 'x-timestamp')
     put_req.environ['wsgi.input'] = FileLikeIter(source_resp.app_iter)
     put_resp = put_req.get_response(self.app)
     close_if_possible(source_resp.app_iter)
     return put_resp

 def _internal_iter(self):
     # Top level of our iterator stack: pass bytes through; catch and
     # handle exceptions.
     try:
         for chunk in self._time_limited_iter():
             yield chunk
     except (ListingIterError, SegmentError) as err:
         self.logger.error(err)
         if not self.validated_first_segment:
             raise
     finally:
         if self.current_resp:
             close_if_possible(self.current_resp.app_iter)

    def _get_source_object(self, req, path_info):
        # make a pre_auth request in case the user has write access
        # to container, but not READ. This was allowed in previous version
        # (i.e., before middleware) so keeping the same behavior here
        get_req = make_pre_authed_request(
            req.environ, path=path_info,
            headers={'X-Newest': 'True'}, method='GET', swift_source='VW')
        source_resp = get_req.get_response(self.app)

        if source_resp.content_length is None or \
                source_resp.content_length > MAX_FILE_SIZE:
            close_if_possible(source_resp.app_iter)
            return HTTPRequestEntityTooLarge(request=req)

        return source_resp

    def handle_obj_versions_put(self, req, versions_cont, api_version,
                                account_name, object_name):
        """
        Copy current version of object to versions_container before proceding
        with original request.

        :param req: original request.
        :param versions_cont: container where previous versions of the object
                              are stored.
        :param api_version: api version.
        :param account_name: account name.
        :param object_name: name of object of original request
        """
        if 'X-Object-Manifest' in req.headers:
            # do not version DLO manifest, proceed with original request
            return self.app

        get_resp = self._get_source_object(req, req.path_info)

        if 'X-Object-Manifest' in get_resp.headers:
            # do not version DLO manifest, proceed with original request
            close_if_possible(get_resp.app_iter)
            return self.app
        if get_resp.status_int == HTTP_NOT_FOUND:
            # nothing to version, proceed with original request
            close_if_possible(get_resp.app_iter)
            return self.app

        # check for any other errors
        self._check_response_error(req, get_resp)

        # if there's an existing object, then copy it to
        # X-Versions-Location
        prefix_len = '%03x' % len(object_name)
        lprefix = prefix_len + object_name + '/'
        ts_source = get_resp.headers.get(
            'x-timestamp',
            calendar.timegm(time.strptime(
                get_resp.headers['last-modified'],
                '%a, %d %b %Y %H:%M:%S GMT')))
        vers_obj_name = lprefix + Timestamp(ts_source).internal

        put_path_info = "/%s/%s/%s/%s" % (
            api_version, account_name, versions_cont, vers_obj_name)
        put_resp = self._put_versioned_obj(req, put_path_info, get_resp)

        self._check_response_error(req, put_resp)
        return self.app

    def __call__(self, env, start_response):
        path = SUB_PUT_POST_PATH
        if env['REQUEST_METHOD'] == 'GET':
            path = SUB_GET_PATH

        # Make a subrequest that will be logged
        hdrs = {'content-type': 'text/plain'}
        sub_req = make_subrequest(env, path=path,
                                  method=self.conf['subrequest_type'],
                                  headers=hdrs,
                                  agent='FakeApp',
                                  swift_source='FA')
        self.register(self.conf['subrequest_type'],
                      path, HTTPOk, headers=hdrs)

        resp = sub_req.get_response(self.app)
        close_if_possible(resp.app_iter)

        return self.app(env, start_response)

    def handle_request(self, req, start_response):
        """
        Take a GET or HEAD request, and if it is for a dynamic large object
        manifest, return an appropriate response.

        Otherwise, simply pass it through.
        """
        resp_iter = self._app_call(req.environ)

        # make sure this response is for a dynamic large object manifest
        for header, value in self._response_headers:
            if (header.lower() == 'x-object-manifest'):
                close_if_possible(resp_iter)
                response = self.get_or_head_response(req, value)
                return response(req.environ, start_response)
        # Not a dynamic large object manifest; just pass it through.
        start_response(self._response_status,
                       self._response_headers,
                       self._response_exc_info)
        return resp_iter

    def _copy_current(self, req, versions_cont, api_version, account_name,
                      object_name):
        # validate the write access to the versioned container before
        # making any backend requests
        if 'swift.authorize' in req.environ:
            container_info = get_container_info(
                req.environ, self.app)
            req.acl = container_info.get('write_acl')
            aresp = req.environ['swift.authorize'](req)
            if aresp:
                raise aresp

        get_resp = self._get_source_object(req, req.path_info)

        if 'X-Object-Manifest' in get_resp.headers:
            # do not version DLO manifest, proceed with original request
            close_if_possible(get_resp.app_iter)
            return
        if get_resp.status_int == HTTP_NOT_FOUND:
            # nothing to version, proceed with original request
            close_if_possible(get_resp.app_iter)
            return

        # check for any other errors
        self._check_response_error(req, get_resp)

        # if there's an existing object, then copy it to
        # X-Versions-Location
        ts_source = get_resp.headers.get(
            'x-timestamp',
            calendar.timegm(time.strptime(
                get_resp.headers['last-modified'],
                '%a, %d %b %Y %H:%M:%S GMT')))
        vers_obj_name = self._build_versions_object_name(
            object_name, ts_source)

        put_path_info = "/%s/%s/%s/%s" % (
            api_version, account_name, versions_cont, vers_obj_name)
        put_resp = self._put_versioned_obj(req, put_path_info, get_resp)

        self._check_response_error(req, put_resp)

    def _restore_data(self, req, versions_cont, api_version, account_name,
                      container_name, object_name, prev_obj_name):
        get_path = "/%s/%s/%s/%s" % (
            api_version, account_name, versions_cont, prev_obj_name)

        get_resp = self._get_source_object(req, get_path)

        # if the version isn't there, keep trying with previous version
        if get_resp.status_int == HTTP_NOT_FOUND:
            close_if_possible(get_resp.app_iter)
            return False

        self._check_response_error(req, get_resp)

        put_path_info = "/%s/%s/%s/%s" % (
            api_version, account_name, container_name, object_name)
        put_resp = self._put_versioned_obj(req, put_path_info, get_resp)

        self._check_response_error(req, put_resp)
        close_if_possible(put_resp.app_iter)
        return get_path

    def handle_obj_versions_delete_push(self, req, versions_cont, api_version,
                                        account_name, container_name,
                                        object_name):
        """
        Handle DELETE requests when in history mode.

        Copy current version of object to versions_container and write a
        delete marker before proceeding with original request.

        :param req: original request.
        :param versions_cont: container where previous versions of the object
                              are stored.
        :param api_version: api version.
        :param account_name: account name.
        :param object_name: name of object of original request
        """
        self._copy_current(req, versions_cont, api_version, account_name,
                           object_name)

        marker_path = "/%s/%s/%s/%s" % (
            api_version, account_name, versions_cont,
            self._build_versions_object_name(object_name, time.time()))
        marker_headers = {
            # Definitive source of truth is Content-Type, and since we add
            # a swift_* param, we know users haven't set it themselves.
            # This is still open to users POSTing to update the content-type
            # but they're just shooting themselves in the foot then.
            'content-type': DELETE_MARKER_CONTENT_TYPE,
            'content-length': '0',
            'x-auth-token': req.headers.get('x-auth-token')}
        marker_req = make_pre_authed_request(
            req.environ, path=marker_path,
            headers=marker_headers, method='PUT', swift_source='VW')
        marker_req.environ['swift.content_type_overridden'] = True
        marker_resp = marker_req.get_response(self.app)
        self._check_response_error(req, marker_resp)
        close_if_possible(marker_resp.app_iter)

        # successfully copied and created delete marker; safe to delete
        return self.app