4

我正在使用 Pyramid 编写一个 Web 应用程序，并希望限制 POST 请求的最大长度，这样人们就不能发布大量数据并耗尽服务器上的所有内存。然而，我几乎在所有我能想到的地方（Pyramid、WebOb、Paster）都找了找，但找不到任何选择来完成这个任务。我已经看到 Paster 对 HTTP 标头的数量、每个标头的长度等有限制，但是对于请求正文的大小，我没有看到任何东西。

服务器将只接受 JSON-RPC 的 POST 请求，因此我不需要允许巨大的请求正文大小。金字塔堆栈中是否有办法实现这一点？

以防万一这在其他方面并不明显，在检查长度并返回 4xx 错误代码之前必须接受整个请求正文并将其加载到内存中的解决方案违背了我正在尝试做的事情的目的，而不是我在找什么。

4

2 回答 2

2

不是真正直接回答您的问题。据我所知，您可以创建一个 wsgi 应用程序，如果主体低于配置设置，则该应用程序将加载请求，您可以将其传递给下一个 WSGI 层。如果超出，您可以停止阅读并直接返回错误。

但老实说，我真的不明白在金字塔中做这件事的意义。例如，如果您使用 nginx 或 apache 或其他东西在反向代理后面运行金字塔。您始终可以限制前端服务器的请求大小。

除非您想在没有任何代理的情况下直接使用 Waitress 或 Paster 运行金字塔，否则您应该在前端服务器中处理应该比 python 更有效的 body size。

编辑

我做了一些研究，这不是一个完整的答案，但我猜这里有一些可以使用的东西。据我所知，您必须阅读 environ['wsgi_input'] 。这是一个类似对象的文件，例如从 nginx 或 apache 接收数据块。

您真正需要做的是读取该文件，直到达到最大长度。如果达到它，如果它不继续请求，则会引发错误。

你可能想看看这个答案

于 2012-04-20T01:25:33.670 回答
1

您可以通过多种方式做到这一点，这里有几个例子。一种使用基于 webob 的 wsgi 中间件（在您安装金字塔时安装）。和一个使用金字塔事件机制的

"""
restricting execution based on request body size
"""
from pyramid.config import Configurator
from pyramid.view import view_config
from pyramid.events import NewRequest, subscriber
from webob import Response, Request
from webob.exc import HTTPBadRequest
import unittest


def restrict_body_middleware(app, max_size=0):
    """
    this is straight wsgi middleware and in this case only depends on
    webob. this can be used with any wsgi compliant web
    framework(which is pretty much all of them)
    """
    def m(environ, start_response):
        r = Request(environ)
        if r.content_length <= max_size:
            return r.get_response(app)(environ, start_response)
        else:
            err_body = """
            request content_length(%s) exceeds
            the configured maximum content_length allowed(%s)
            """ % (r.content_length, max_size)
            res = HTTPBadRequest(err_body)
            return res(environ, start_response)

    return m


def new_request_restrict(event):
    """
    pyramid event handler called whenever there is a new request
    recieved

    http://docs.pylonsproject.org/projects/pyramid/en/1.2-branch/narr/events.html
    """
    request = event.request
    if request.content_length >= 0:
        raise HTTPBadRequest("too big")


@view_config()
def index(request):
    return Response("HI THERE")


def make_application():
    """
    make appplication with one view
    """
    config = Configurator()
    config.scan()
    return config.make_wsgi_app()


def make_application_with_event():
    """
    make application with one view and one event subsriber subscribed
    to NewRequest
    """
    config = Configurator()
    config.add_subscriber(new_request_restrict, NewRequest)
    return config.make_wsgi_app()


def make_application_with_middleware():
    """
    make application with one view wrapped in wsgi middleware
    """
    return restrict_body_middleware(make_application())



class TestWSGIApplication(unittest.TestCase):
    def testNoRestriction(self):
        app = make_application()
        request = Request.blank("/", body="i am a request with a body")
        self.assert_(request.content_length > 0, "content_length should be > 0")
        response = request.get_response(app)
        self.assert_(response.status_int == 200, "expected status code 200 got %s" % response.status_int)

    def testRestrictedByMiddleware(self):
        app = make_application_with_middleware()
        request = Request.blank("/", body="i am a request with a body")
        self.assert_(request.content_length > 0, "content_length should be > 0")
        response = request.get_response(app)
        self.assert_(response.status_int == 400, "expected status code 400 got %s" % response.status_int)

    def testRestrictedByEvent(self):
        app = make_application_with_event()
        request = Request.blank("/", body="i am a request with a body")
        self.assert_(request.content_length > 0, "content_length should be > 0")
        response = request.get_response(app)
        self.assert_(response.status_int == 400, "expected status code 400 got %s" % response.status_int)



if __name__ == "__main__":
    unittest.main()
于 2012-04-20T11:58:34.060 回答