diff --git a/tests/unit/forklift/test_legacy.py b/tests/unit/forklift/test_legacy.py index 6d25b80594f5..b6ccb10bf035 100644 --- a/tests/unit/forklift/test_legacy.py +++ b/tests/unit/forklift/test_legacy.py @@ -753,6 +753,25 @@ def test_is_duplicate_false(self, pyramid_config, db_request): class TestFileUpload: + def test_fails_disallow_new_upload(self, pyramid_config, pyramid_request): + pyramid_config.testing_securitypolicy(userid=1) + pyramid_request.flags = pretend.stub( + enabled=lambda value: value == AdminFlagValue.DISALLOW_NEW_UPLOAD + ) + pyramid_request.help_url = pretend.call_recorder(lambda **kw: "/the/help/url/") + pyramid_request.user = pretend.stub(primary_email=pretend.stub(verified=True)) + + with pytest.raises(HTTPForbidden) as excinfo: + legacy.file_upload(pyramid_request) + + resp = excinfo.value + + assert resp.status_code == 403 + assert resp.status == ( + "403 New uploads are temporarily disabled. " + "See /the/help/url/ for details" + ) + @pytest.mark.parametrize("version", ["2", "3", "-1", "0", "dog", "cat"]) def test_fails_invalid_version(self, pyramid_config, pyramid_request, version): pyramid_config.testing_securitypolicy(userid=1) diff --git a/warehouse/admin/flags.py b/warehouse/admin/flags.py index f3c580e7e6a0..fc89ea0705ea 100644 --- a/warehouse/admin/flags.py +++ b/warehouse/admin/flags.py @@ -18,6 +18,7 @@ class AdminFlagValue: DISALLOW_DELETION = "disallow-deletion" DISALLOW_NEW_PROJECT_REGISTRATION = "disallow-new-project-registration" + DISALLOW_NEW_UPLOAD = "disallow-new-upload" DISALLOW_NEW_USER_REGISTRATION = "disallow-new-user-registration" READ_ONLY = "read-only" diff --git a/warehouse/forklift/legacy.py b/warehouse/forklift/legacy.py index c3771cc5080a..a009a30ac555 100644 --- a/warehouse/forklift/legacy.py +++ b/warehouse/forklift/legacy.py @@ -734,6 +734,15 @@ def file_upload(request): HTTPForbidden, "Read-only mode: Uploads are temporarily disabled" ) + if request.flags.enabled(AdminFlagValue.DISALLOW_NEW_UPLOAD): + raise _exc_with_message( + HTTPForbidden, + "New uploads are temporarily disabled. " + "See {projecthelp} for details".format( + projecthelp=request.help_url(_anchor="admin-intervention") + ), + ) + # Log an attempt to upload metrics = request.find_service(IMetricsService, context=None) metrics.increment("warehouse.upload.attempt") diff --git a/warehouse/migrations/versions/ee4c59b2ef3a_add_disallow_new_upload_adminflag.py b/warehouse/migrations/versions/ee4c59b2ef3a_add_disallow_new_upload_adminflag.py new file mode 100644 index 000000000000..97e4d0bc7603 --- /dev/null +++ b/warehouse/migrations/versions/ee4c59b2ef3a_add_disallow_new_upload_adminflag.py @@ -0,0 +1,41 @@ +# 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. +""" +Add disallow-new-upload AdminFlag + +Revision ID: ee4c59b2ef3a +Revises: 8650482fb903 +Create Date: 2019-08-23 22:34:29.180163 +""" + +from alembic import op + +revision = "ee4c59b2ef3a" +down_revision = "8650482fb903" + + +def upgrade(): + op.execute( + """ + INSERT INTO admin_flags(id, description, enabled, notify) + VALUES ( + 'disallow-new-upload', + 'Disallow ALL new uploads', + FALSE, + FALSE + ) + """ + ) + + +def downgrade(): + op.execute("DELETE FROM admin_flags WHERE id = 'disallow-new-upload'")