Fix invalid boundary (pypi#4572)

Author: dstufft

tests/unit/test_config.py

@@ -16,79 +16,13 @@
 import pytest
 
 from pyramid import renderers
-from pyramid.request import Request
 from pyramid.tweens import EXCVIEW
 
 from warehouse import config
 from warehouse.errors import BasicAuthBreachedPassword
 from warehouse.utils.wsgi import ProxyFixer, VhmRootRemover, HostRewrite
 
 
-class TestJunkEncodingTween:
-    def test_valid(self):
-        response = pretend.stub()
-        handler = pretend.call_recorder(lambda request: response)
-
-        tween = config.junk_encoding_tween_factory(handler, pretend.stub())
-
-        request = Request({"QUERY_STRING": ":action=browse", "PATH_INFO": "/pypi"})
-        resp = tween(request)
-
-        assert resp is response
-
-    def test_invalid_qsl(self):
-        response = pretend.stub()
-        handler = pretend.call_recorder(lambda request: response)
-
-        tween = config.junk_encoding_tween_factory(handler, pretend.stub())
-
-        request = Request({"QUERY_STRING": "%Aaction=browse"})
-        resp = tween(request)
-
-        assert resp is not response
-        assert resp.status_code == 400
-        assert resp.detail == "Invalid bytes in query string."
-
-    def test_invalid_path(self):
-        response = pretend.stub()
-        handler = pretend.call_recorder(lambda request: response)
-
-        tween = config.junk_encoding_tween_factory(handler, pretend.stub())
-
-        request = Request({"PATH_INFO": "/projects/abouÅt"})
-        resp = tween(request)
-
-        assert resp is not response
-        assert resp.status_code == 400
-        assert resp.detail == "Invalid bytes in URL."
-
-
-class TestUnicodeRedirectTween:
-    def test_basic_redirect(self):
-        response = pretend.stub(location="/a/path/to/nowhere")
-        handler = pretend.call_recorder(lambda request: response)
-        registry = pretend.stub()
-        tween = config.unicode_redirect_tween_factory(handler, registry)
-        request = pretend.stub(path="/A/pAtH/tO/nOwHeRe/")
-        assert tween(request) == response
-
-    def test_unicode_basic_redirect(self):
-        response = pretend.stub(location="/pypi/\u2603/json/")
-        handler = pretend.call_recorder(lambda request: response)
-        registry = pretend.stub()
-        tween = config.unicode_redirect_tween_factory(handler, registry)
-        request = pretend.stub(path="/pypi/snowman/json/")
-        assert tween(request).location == "/pypi/%E2%98%83/json/"
-
-    def test_not_redirect(self):
-        response = pretend.stub(location=None)
-        handler = pretend.call_recorder(lambda request: response)
-        registry = pretend.stub()
-        tween = config.unicode_redirect_tween_factory(handler, registry)
-        request = pretend.stub(path="/wu/tang/")
-        assert tween(request) == response
-
-
 class TestRequireHTTPSTween:
     def test_noops_when_disabled(self):
         handler = pretend.stub()
@@ -393,6 +327,7 @@ def __init__(self):
             pretend.call(".http"),
         ]
         + [pretend.call(x) for x in [configurator_settings.get("warehouse.theme")] if x]
+        + [pretend.call(".sanity")]
     )
     assert configurator_obj.add_jinja2_renderer.calls == [
         pretend.call(".html"),
@@ -420,22 +355,6 @@ def __init__(self):
     add_settings_dict = configurator_obj.add_settings.calls[2].args[0]
     assert add_settings_dict["tm.manager_hook"](pretend.stub()) is transaction_manager
     assert configurator_obj.add_tween.calls == [
-        pretend.call(
-            "warehouse.config.junk_encoding_tween_factory",
-            over=[
-                "warehouse.referrer_policy.referrer_policy_tween_factory",
-                "warehouse.config.require_https_tween_factory",
-                "warehouse.config.unicode_redirect_tween_factory",
-                "warehouse.csp.content_security_policy_tween_factory",
-                "warehouse.static.whitenoise_tween_factory",
-                "warehouse.utils.compression.compression_tween_factory",
-                "warehouse.raven.raven_tween_factory",
-                "pyramid_tm.tm_tween_factory",
-                "pyramid.tweens.excview_tween_factory",
-                "warehouse.cache.http.conditional_http_tween_factory",
-            ],
-        ),
-        pretend.call("warehouse.config.unicode_redirect_tween_factory"),
         pretend.call("warehouse.config.require_https_tween_factory"),
         pretend.call(
             "warehouse.utils.compression.compression_tween_factory",

tests/unit/test_sanity.py

@@ -0,0 +1,154 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import io
+
+import pretend
+import pytest
+
+from pyramid.httpexceptions import HTTPMovedPermanently, HTTPBadRequest
+from pyramid.request import Request
+from pyramid.response import Response
+
+from warehouse import sanity
+
+
+class TestJunkEncoding:
+    def test_valid(self):
+        request = Request({"QUERY_STRING": ":action=browse", "PATH_INFO": "/pypi"})
+        sanity.junk_encoding(request)
+
+    def test_invalid_qsl(self):
+        request = Request({"QUERY_STRING": "%Aaction=browse"})
+
+        with pytest.raises(HTTPBadRequest, match="Invalid bytes in query string."):
+            sanity.junk_encoding(request)
+
+    def test_invalid_path(self):
+        request = Request({"PATH_INFO": "/projects/abouÅt"})
+
+        with pytest.raises(HTTPBadRequest, match="Invalid bytes in URL."):
+            sanity.junk_encoding(request)
+
+
+class TestInvalidForms:
+    def test_valid(self):
+        request = Request(
+            {
+                "REQUEST_METHOD": "POST",
+                "CONTENT_TYPE": (
+                    "multipart/form-data; boundary=c397e2aa2980f1a53dee37c05b8fb45a"
+                ),
+                "wsgi.input": io.BytesIO(
+                    b"--------------------------c397e2aa2980f1a53dee37c05b8fb45a\r\n"
+                    b'Content-Disposition: form-data; name="person"\r\n'
+                    b"anonymous"
+                ),
+            }
+        )
+
+        sanity.invalid_forms(request)
+
+    def test_invalid_form(self):
+        request = Request(
+            {
+                "REQUEST_METHOD": "POST",
+                "CONTENT_TYPE": ("multipart/form-data"),
+                "wsgi.input": io.BytesIO(
+                    b'Content-Disposition: form-data; name="person"\r\n' b"anonymous"
+                ),
+            }
+        )
+
+        with pytest.raises(HTTPBadRequest, match="Invalid Form Data."):
+            sanity.invalid_forms(request)
+
+    def test_not_post(self):
+        request = Request({"REQUEST_METHOD": "GET"})
+        sanity.invalid_forms(request)
+
+
+@pytest.mark.parametrize(
+    ("original_location", "expected_location"),
+    [
+        ("/a/path/to/nowhere", "/a/path/to/nowhere"),
+        ("/project/☃/", "/project/%E2%98%83/"),
+        (None, None),
+    ],
+)
+def test_unicode_redirects(original_location, expected_location):
+    if original_location:
+        resp_in = HTTPMovedPermanently(original_location)
+    else:
+        resp_in = Response()
+
+    resp_out = sanity.unicode_redirects(resp_in)
+
+    assert resp_out.location == expected_location
+
+
+class TestSanityTween:
+    def test_ingress_valid(self, monkeypatch):
+        junk_encoding = pretend.call_recorder(lambda request: None)
+        monkeypatch.setattr(sanity, "junk_encoding", junk_encoding)
+
+        invalid_forms = pretend.call_recorder(lambda request: None)
+        monkeypatch.setattr(sanity, "invalid_forms", invalid_forms)
+
+        response = pretend.stub()
+        handler = pretend.call_recorder(lambda request: response)
+        registry = pretend.stub()
+
+        request = pretend.stub()
+
+        tween = sanity.sanity_tween_factory_ingress(handler, registry)
+
+        assert tween(request) is response
+        assert junk_encoding.calls == [pretend.call(request)]
+        assert invalid_forms.calls == [pretend.call(request)]
+        assert handler.calls == [pretend.call(request)]
+
+    def test_ingress_invalid(self, monkeypatch):
+        response = HTTPBadRequest()
+
+        @pretend.call_recorder
+        def junk_encoding(request):
+            raise response
+
+        monkeypatch.setattr(sanity, "junk_encoding", junk_encoding)
+
+        handler = pretend.call_recorder(lambda request: response)
+        registry = pretend.stub()
+
+        request = pretend.stub()
+
+        tween = sanity.sanity_tween_factory_ingress(handler, registry)
+
+        assert tween(request) is response
+        assert junk_encoding.calls == [pretend.call(request)]
+        assert handler.calls == []
+
+    def test_egress(self, monkeypatch):
+        unicode_redirects = pretend.call_recorder(lambda resp: resp)
+        monkeypatch.setattr(sanity, "unicode_redirects", unicode_redirects)
+
+        response = pretend.stub()
+        handler = pretend.call_recorder(lambda request: response)
+        registry = pretend.stub()
+
+        request = pretend.stub()
+
+        tween = sanity.sanity_tween_factory_egress(handler, registry)
+
+        assert tween(request) is response
+        assert handler.calls == [pretend.call(request)]
+        assert unicode_redirects.calls == [pretend.call(response)]

warehouse/config.py

@@ -13,13 +13,11 @@
 import enum
 import os
 import shlex
-import urllib.parse
 
 import transaction
 
 from pyramid import renderers
 from pyramid.config import Configurator as _Configurator
-from pyramid.httpexceptions import HTTPBadRequest
 from pyramid.response import Response
 from pyramid.security import Allow, Authenticated
 from pyramid.tweens import EXCVIEW
@@ -63,47 +61,6 @@ def __init__(self, request):
         pass
 
 
-def junk_encoding_tween_factory(handler, request):
-    def junk_encoding_tween(request):
-        # We're going to test our request a bit, before we pass it into the
-        # handler. This will let us return a better error than a 500 if we
-        # can't decode these.
-
-        # Ref: https://github.com/Pylons/webob/issues/161
-        # Ref: https://github.com/Pylons/webob/issues/115
-        try:
-            request.GET.get("", None)
-        except UnicodeDecodeError:
-            return HTTPBadRequest("Invalid bytes in query string.")
-
-        # Look for invalid bytes in a path.
-        try:
-            request.path_info
-        except UnicodeDecodeError:
-            return HTTPBadRequest("Invalid bytes in URL.")
-
-        # Everything worked! Handle this request as normal.
-        return handler(request)
-
-    return junk_encoding_tween
-
-
-def unicode_redirect_tween_factory(handler, request):
-    def unicode_redirect_tween(request):
-        response = handler(request)
-        if response.location:
-            try:
-                response.location.encode("ascii")
-            except UnicodeEncodeError:
-                response.location = "/".join(
-                    [urllib.parse.quote_plus(x) for x in response.location.split("/")]
-                )
-
-        return response
-
-    return unicode_redirect_tween
-
-
 def require_https_tween_factory(handler, registry):
 
     if not registry.settings.get("enforce_https", True):
@@ -265,24 +222,6 @@ def configure(settings=None):
     config = Configurator(settings=settings)
     config.set_root_factory(RootFactory)
 
-    # Add some fixups for some encoding/decoding issues
-    config.add_tween(
-        "warehouse.config.junk_encoding_tween_factory",
-        over=[
-            "warehouse.referrer_policy.referrer_policy_tween_factory",
-            "warehouse.config.require_https_tween_factory",
-            "warehouse.config.unicode_redirect_tween_factory",
-            "warehouse.csp.content_security_policy_tween_factory",
-            "warehouse.static.whitenoise_tween_factory",
-            "warehouse.utils.compression.compression_tween_factory",
-            "warehouse.raven.raven_tween_factory",
-            "pyramid_tm.tm_tween_factory",
-            "pyramid.tweens.excview_tween_factory",
-            "warehouse.cache.http.conditional_http_tween_factory",
-        ],
-    )
-    config.add_tween("warehouse.config.unicode_redirect_tween_factory")
-
     # Register support for services
     config.include("pyramid_services")
 
@@ -527,6 +466,11 @@ def configure(settings=None):
         ignore=["warehouse.migrations.env", "warehouse.celery", "warehouse.wsgi"]
     )
 
+    # Sanity check our request and responses.
+    # Note: It is very important that this go last. We need everything else that might
+    #       have added a tween to be registered prior to this.
+    config.include(".sanity")
+
     # Finally, commit all of our changes
     config.commit()