4

Pyramidを使用してWebアプリケーションを作成していますが、POSTリクエストの最大長を制限して、大量のデータを投稿したり、サーバー上のすべてのメモリを使い果たしたりしないようにしたいと考えています。ただし、考えられるすべての場所（Pyramid、WebOb、Paster）を調べたところ、これを実現するためのオプションが見つかりませんでした。PasterにはHTTPヘッダーの数、各ヘッダーの長さなどに制限があることを確認しましたが、リクエスト本文のサイズについては何も表示されませんでした。

サーバーはJSON-RPCに対してのみPOSTリクエストを受け入れるため、リクエストの本文のサイズを大きくする必要はありません。これを達成する方法はピラミッドスタックにありますか？

これが残りの部分から明らかでない場合に備えて、長さをチェックして4xxエラーコードを返す前にリクエスト本文全体を受け入れてメモリにロードする必要があるソリューションは、私がやろうとしていることの目的を無効にしますが、そうではありません私が探しているもの。

4

2 に答える 2

2

あなたの質問に対する直接の答えではありません。私の知る限り、本体が構成設定を下回っている場合にリクエストを読み込むwsgiアプリを作成して、次のWSGIレイヤーに渡すことができます。それが上に行く場合は、読み取りを停止してエラーを直接返すことができます。

しかし、正直なところ、ピラミッドでそれを行う意味は本当にわかりません。たとえば、nginxやapacheなどを使用してリバースプロキシの背後でピラミッドを実行する場合、フロントエンドサーバーでリクエストのサイズをいつでも制限できます。

プロキシなしでウェイトレスまたはパスターを使用してピラミッドを直接実行する場合を除き、フロントエンドサーバーでPythonよりも効率的なボディサイズを処理する必要があります。

編集

私はいくつかの調査をしました、それは完全な答えではありません、しかしここに私が推測することができる何かがあります。私が知る限り、environ['wsgi_input']を読む必要があります。これは、たとえばnginxやapacheからデータのチャンクを受け取るオブジェクトのようなファイルです。

あなたが本当にしなければならないことは、最大長に達するまでそのファイルを読むことです。到達した場合、要求を続行しないとエラーが発生します。

あなたはこの答えを見たいかもしれません

于 2012-04-20T01:25:33.670 に答える
1

さまざまな方法でそれを行うことができます。ここにいくつかの例を示します。1つはwebobに基づくwsgiミドルウェアを使用します（特にpyramidをインストールするときにインストールされます）。ピラミッドイベントメカニズムを使用するもの

"""
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 に答える