Implement an Admin Flag model and some Emergency Circuit breakers

tests/common/db/utils.py

@@ -0,0 +1,26 @@
+# 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 factory.fuzzy
+
+from warehouse.utils.admin_flags import AdminFlag
+
+from .base import WarehouseFactory
+
+
+class AdminFlagFactory(WarehouseFactory):
+    class Meta:
+        model = AdminFlag
+
+    id = factory.fuzzy.FuzzyText(length=12)
+    description = factory.fuzzy.FuzzyText(length=24)
+    enabled = True

tests/unit/accounts/test_views.py

@@ -17,14 +17,15 @@
 import pretend
 import pytest
 
-from pyramid.httpexceptions import HTTPMovedPermanently, HTTPSeeOther
+from pyramid.httpexceptions import (HTTPMovedPermanently, HTTPSeeOther)
 from sqlalchemy.orm.exc import NoResultFound
 
 from warehouse.accounts import views
 from warehouse.accounts.interfaces import (
     IUserService, ITokenService, TokenExpired, TokenInvalid, TokenMissing,
     TooManyFailedLogins
 )
+from warehouse.utils.admin_flags import AdminFlag
 
 from ...common.db.accounts import EmailFactory, UserFactory
 
@@ -288,32 +289,32 @@ def test_post_redirects_user(self, pyramid_request, expected_next_url,
 
 class TestRegister:
 
-    def test_get(self, pyramid_request):
+    def test_get(self, db_request):
         form_inst = pretend.stub()
         form = pretend.call_recorder(lambda *args, **kwargs: form_inst)
-        pyramid_request.find_service = pretend.call_recorder(
+        db_request.find_service = pretend.call_recorder(
             lambda *args, **kwargs: pretend.stub(
                 enabled=False,
                 csp_policy=pretend.stub(),
                 merge=lambda _: None,
             )
         )
-        result = views.register(pyramid_request, _form_class=form)
+        result = views.register(db_request, _form_class=form)
         assert result["form"] is form_inst
 
     def test_redirect_authenticated_user(self):
         result = views.register(pretend.stub(authenticated_userid=1))
         assert isinstance(result, HTTPSeeOther)
         assert result.headers["Location"] == "/"
 
-    def test_register_redirect(self, pyramid_request, monkeypatch):
-        pyramid_request.method = "POST"
+    def test_register_redirect(self, db_request, monkeypatch):
+        db_request.method = "POST"
 
         user = pretend.stub(id=pretend.stub())
         email = pretend.stub()
         create_user = pretend.call_recorder(lambda *args, **kwargs: user)
         add_email = pretend.call_recorder(lambda *args, **kwargs: email)
-        pyramid_request.find_service = pretend.call_recorder(
+        db_request.find_service = pretend.call_recorder(
             lambda *args, **kwargs: pretend.stub(
                 csp_policy={},
                 merge=lambda _: {},
@@ -326,8 +327,8 @@ def test_register_redirect(self, pyramid_request, monkeypatch):
                 add_email=add_email,
             )
         )
-        pyramid_request.route_path = pretend.call_recorder(lambda name: "/")
-        pyramid_request.POST.update({
+        db_request.route_path = pretend.call_recorder(lambda name: "/")
+        db_request.POST.update({
             "username": "username_value",
             "password": "MyStr0ng!shP455w0rd",
             "password_confirm": "MyStr0ng!shP455w0rd",
@@ -337,7 +338,7 @@ def test_register_redirect(self, pyramid_request, monkeypatch):
         send_email = pretend.call_recorder(lambda *a: None)
         monkeypatch.setattr(views, 'send_email_verification_email', send_email)
 
-        result = views.register(pyramid_request)
+        result = views.register(db_request)
 
         assert isinstance(result, HTTPSeeOther)
         assert result.headers["Location"] == "/"
@@ -347,7 +348,40 @@ def test_register_redirect(self, pyramid_request, monkeypatch):
         assert add_email.calls == [
             pretend.call(user.id, '[email protected]', primary=True),
         ]
-        assert send_email.calls == [pretend.call(pyramid_request, email)]
+        assert send_email.calls == [pretend.call(db_request, email)]
+
+    def test_register_fails_with_admin_flag_set(self, db_request):
+        admin_flag = (db_request.db.query(AdminFlag)
+                      .filter(
+                          AdminFlag.id == 'disallow-new-user-registration')
+                      .first())
+        admin_flag.enabled = True
+        db_request.method = "POST"
+
+        db_request.POST.update({
+            "username": "username_value",
+            "password": "MyStr0ng!shP455w0rd",
+            "password_confirm": "MyStr0ng!shP455w0rd",
+            "email": "[email protected]",
+            "full_name": "full_name",
+        })
+
+        db_request.session.flash = pretend.call_recorder(
+            lambda *a, **kw: None
+        )
+
+        db_request.route_path = pretend.call_recorder(lambda name: "/")
+
+        result = views.register(db_request)
+
+        assert isinstance(result, HTTPSeeOther)
+        assert db_request.session.flash.calls == [
+            pretend.call(
+                ("New User Registration Temporarily Disabled "
+                 "See https://pypi.org/help#admin-intervention for details"),
+                queue="error"
+            ),
+        ]
 
 
 class TestRequestPasswordReset:

tests/unit/forklift/test_legacy.py

@@ -36,6 +36,7 @@
     File, Filename, Dependency, DependencyKind, Release, Project, Role,
     JournalEntry,
 )
+from warehouse.utils.admin_flags import AdminFlag
 
 from ...common.db.accounts import UserFactory, EmailFactory
 from ...common.db.packaging import (
@@ -924,6 +925,38 @@ def test_fails_with_stdlib_names(self, pyramid_config, db_request, name):
                                 "See https://pypi.org/help/#project-name "
                                 "for more information.").format(name))
 
+    def test_fails_with_admin_flag_set(self, pyramid_config, db_request):
+        admin_flag = (db_request.db.query(AdminFlag)
+                      .filter(
+                          AdminFlag.id == 'disallow-new-project-registration')
+                      .first())
+        admin_flag.enabled = True
+        pyramid_config.testing_securitypolicy(userid=1)
+        name = 'fails-with-admin-flag'
+        db_request.POST = MultiDict({
+            "metadata_version": "1.2",
+            "name": name,
+            "version": "1.0",
+            "filetype": "sdist",
+            "md5_digest": "a fake md5 digest",
+            "content": pretend.stub(
+                filename=f"{name}-1.0.tar.gz",
+                file=io.BytesIO(b"A fake file."),
+                type="application/tar",
+            ),
+        })
+
+        with pytest.raises(HTTPForbidden) as excinfo:
+            legacy.file_upload(db_request)
+
+        resp = excinfo.value
+
+        assert resp.status_code == 403
+        assert resp.status == ("403 New Project Registration Temporarily "
+                               "Disabled See "
+                               "https://pypi.org/help#admin-intervention for "
+                               "details")
+
     def test_upload_fails_without_file(self, pyramid_config, db_request):
         pyramid_config.testing_securitypolicy(userid=1)
         db_request.POST = MultiDict({

tests/unit/utils/test_admin_flags.py

@@ -0,0 +1,25 @@
+# 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.
+
+from warehouse.utils.admin_flags import AdminFlag
+
+from ...common.db.utils import AdminFlagFactory as DBAdminFlagFactory
+
+
+class TestAdminFlag:
+
+    def test_default(self, db_session):
+        assert not AdminFlag.is_enabled(db_session, 'not-a-real-flag')
+
+    def test_enabled(self, db_session):
+        DBAdminFlagFactory.create(id='this-flag-is-enabled', enabled=True)
+        assert AdminFlag.is_enabled(db_session, 'this-flag-is-enabled')

warehouse/accounts/views.py

@@ -15,7 +15,7 @@
 import uuid
 
 from pyramid.httpexceptions import (
-    HTTPMovedPermanently, HTTPSeeOther, HTTPTooManyRequests,
+    HTTPMovedPermanently, HTTPSeeOther, HTTPTooManyRequests
 )
 from pyramid.security import Authenticated, remember, forget
 from pyramid.view import view_config
@@ -36,6 +36,7 @@
     send_password_reset_email, send_email_verification_email,
 )
 from warehouse.packaging.models import Project, Release
+from warehouse.utils.admin_flags import AdminFlag
 from warehouse.utils.http import is_safe_url
 
 
@@ -215,6 +216,14 @@ def register(request, _form_class=RegistrationForm):
     if request.authenticated_userid is not None:
         return HTTPSeeOther("/")
 
+    if AdminFlag.is_enabled(request.db, 'disallow-new-user-registration'):
+        request.session.flash(
+            ("New User Registration Temporarily Disabled "
+             "See https://pypi.org/help#admin-intervention for details"),
+            queue="error",
+        )
+        return HTTPSeeOther(request.route_path("index"))
+
     user_service = request.find_service(IUserService, context=None)
     recaptcha_service = request.find_service(name="recaptcha")
     request.find_service(name="csp").merge(recaptcha_service.csp_policy)

warehouse/forklift/legacy.py

@@ -44,6 +44,7 @@
     Project, Release, Dependency, DependencyKind, Role, File, Filename,
     JournalEntry, BlacklistedProject,
 )
+from warehouse.utils.admin_flags import AdminFlag
 from warehouse.utils import http
 
 
@@ -740,6 +741,18 @@ def file_upload(request):
                           func.normalize_pep426_name(form.name.data)).one()
         )
     except NoResultFound:
+        # Check for AdminFlag set by a PyPI Administrator disabling new project
+        # registration, reasons for this include Spammers, security
+        # vulnerabilities, or just wanting to be lazy and not worry ;)
+        if AdminFlag.is_enabled(
+                request.db,
+                'disallow-new-project-registration'):
+            raise _exc_with_message(
+                HTTPForbidden,
+                ("New Project Registration Temporarily Disabled "
+                 "See https://pypi.org/help#admin-intervention for details"),
+            ) from None
+
         # Ensure that user has at least one verified email address. This should
         # reduce the ease of spam account creation and activity.
         # TODO: Once legacy is shutdown consider the condition here, perhaps

warehouse/migrations/versions/7165e957cddc_create_table_for_warehouse_.py

@@ -0,0 +1,56 @@
+# 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.
+"""
+create table for warehouse administration flags
+
+Revision ID: 7165e957cddc
+Revises: 1e2ccd34f539
+Create Date: 2018-02-17 18:42:18.209572
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = '7165e957cddc'
+down_revision = '1e2ccd34f539'
+
+
+def upgrade():
+    op.create_table(
+        'warehouse_admin_flag',
+        sa.Column('id', sa.Text(), nullable=False),
+        sa.Column('description', sa.Text(), nullable=False),
+        sa.Column('enabled', sa.Boolean(), nullable=False),
+        sa.PrimaryKeyConstraint('id')
+    )
+    # Insert our initial flags.
+    op.execute("""
+        INSERT INTO warehouse_admin_flag(id, description, enabled)
+        VALUES (
+            'disallow-new-user-registration',
+            'Disallow ALL new User registrations',
+            FALSE
+        )
+    """)
+    op.execute("""
+        INSERT INTO warehouse_admin_flag(id, description, enabled)
+        VALUES (
+            'disallow-new-project-registration',
+            'Disallow ALL new Project registrations',
+            FALSE
+        )
+    """)
+
+
+def downgrade():
+    op.drop_table('warehouse_admin_flag')

warehouse/templates/pages/help.html

@@ -36,6 +36,7 @@
 {% macro availability() %}Can I depend on PyPI being available?{% endmacro %}
 {% macro mirroring() %}How can I run a mirror of PyPI?{% endmacro %}
 {% macro private_indices() %}How can I publish my private packages to PyPI?{% endmacro %}
+{% macro admin_intervention() %}Why did my package or user registration get blocked?{% endmacro %}
 
 {% block title %}Help{% endblock %}
 
@@ -61,6 +62,7 @@ <h1 class="page-title">Common Questions</h1>
         <li><a href="#availability">{{ availability() }}</a></li>
         <li><a href="#mirroring">{{ mirroring() }}</a></li>
         <li><a href="#private-indices">{{ private_indices() }}</a></li>
+        <li><a href="#admin-intervention">{{ admin_intervention() }}</a></li>
       </ul>
 
       <section id="packages" class="common-question">
@@ -256,6 +258,16 @@ <h2>{{ private_indices() }}</h2>
           PyPI does not support publishing private packages. If you need to publish your private package to a package index, the recommended solution is to run your own deployment of the <a href="https://pypi.org/project/devpi/">devpi project</a>.
         </p>
       </section>
+
+      <section id="admin-intervention" class="common-question">
+        <h2>{{ admin_intervention() }}</h2>
+        <p>
+          Spammers return to PyPI on some regularity hoping to place their Search Engine Optimized phishing, scam, and click-farming content on the site. Since PyPI allows for indexing of the Long Description and other data related to projects and has a generally solid search reputation it is a prime target.
+        </p>
+        <p>
+          When the PyPI Administrators are overwhelmed by spam <b>or</b> determine that there is some other threat to PyPI, new user registration and/or new project registration may be disabled. Check <a href="https://status.python.org">our status page</a> for more details, as we'll likely have updated it with reasoning for the intervention.
+        </p>
+      </section>
     </div>
   </section>