Revert "Revert "models/infra for ip_address tracking and bans"" (#12534)

* Revert "Revert "models/infra for ip_address tracking and bans (#12513)" (#12533)"

This reverts commit 09e934c.

* Set our own request._unauthenticated_userid

* Tests

Co-authored-by: Dustin Ingram <[email protected]>

Author: ewdurbin

tests/common/db/ip_addresses.py

@@ -0,0 +1,20 @@
+# 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.ip_addresses.models import IpAddress
+
+from .base import WarehouseFactory
+
+
+class IpAddressFactory(WarehouseFactory):
+    class Meta:
+        model = IpAddress

tests/conftest.py

@@ -156,6 +156,7 @@ def pyramid_request(pyramid_services, jinja, remote_addr):
     dummy_request.find_service = pyramid_services.find_service
     dummy_request.remote_addr = remote_addr
     dummy_request.authentication_method = pretend.stub()
+    dummy_request._unauthenticated_userid = None
 
     dummy_request.registry.registerUtility(jinja, IJinja2Environment, name=".jinja2")
 
@@ -391,6 +392,7 @@ def query_recorder(app_config):
 def db_request(pyramid_request, db_session):
     pyramid_request.db = db_session
     pyramid_request.flags = admin.flags.Flags(pyramid_request)
+    pyramid_request.banned = admin.bans.Bans(pyramid_request)
     return pyramid_request
 
 

tests/unit/accounts/test_core.py

@@ -321,6 +321,12 @@ def test_without_identity(self):
         assert accounts._user(request) is None
 
 
+class TestUnauthenticatedUserid:
+    def test_unauthenticated_userid(self):
+        request = pretend.stub()
+        assert accounts._unauthenticated_userid(request) is None
+
+
 def test_includeme(monkeypatch):
     authz_obj = pretend.stub()
     authz_cls = pretend.call_recorder(lambda *a, **kw: authz_obj)
@@ -358,7 +364,7 @@ def test_includeme(monkeypatch):
         register_service_factory=pretend.call_recorder(
             lambda factory, iface, name=None: None
         ),
-        add_request_method=pretend.call_recorder(lambda f, name, reify: None),
+        add_request_method=pretend.call_recorder(lambda f, name, reify=False: None),
         set_security_policy=pretend.call_recorder(lambda p: None),
         maybe_dotted=pretend.call_recorder(lambda path: path),
         add_route_predicate=pretend.call_recorder(lambda name, cls: None),
@@ -389,7 +395,8 @@ def test_includeme(monkeypatch):
         pretend.call(RateLimit("3 per 6 hours"), IRateLimiter, name="email.verify"),
     ]
     assert config.add_request_method.calls == [
-        pretend.call(accounts._user, name="user", reify=True)
+        pretend.call(accounts._user, name="user", reify=True),
+        pretend.call(accounts._unauthenticated_userid, name="_unauthenticated_userid"),
     ]
     assert config.set_security_policy.calls == [pretend.call(multi_policy_obj)]
     assert multi_policy_cls.calls == [

tests/unit/accounts/test_forms.py

@@ -71,7 +71,12 @@ def test_validate_username_with_user(self):
         assert user_service.find_userid.calls == [pretend.call("my_username")]
 
     def test_validate_password_no_user(self):
-        request = pretend.stub()
+        request = pretend.stub(
+            remote_addr="1.2.3.4",
+            banned=pretend.stub(
+                by_ip=lambda ip_address: False,
+            ),
+        )
         user_service = pretend.stub(
             find_userid=pretend.call_recorder(lambda userid: None)
         )
@@ -92,7 +97,9 @@ def test_validate_password_no_user(self):
         ]
 
     def test_validate_password_disabled_for_compromised_pw(self, db_session):
-        request = pretend.stub()
+        request = pretend.stub(
+            remote_addr="1.2.3.4", banned=pretend.stub(by_ip=lambda ip_address: False)
+        )
         user_service = pretend.stub(
             find_userid=pretend.call_recorder(lambda userid: 1),
             is_disabled=pretend.call_recorder(
@@ -115,7 +122,12 @@ def test_validate_password_disabled_for_compromised_pw(self, db_session):
         assert user_service.is_disabled.calls == [pretend.call(1)]
 
     def test_validate_password_ok(self):
-        request = pretend.stub(remote_addr="1.2.3.4")
+        request = pretend.stub(
+            remote_addr="1.2.3.4",
+            banned=pretend.stub(
+                by_ip=lambda ip_address: False,
+            ),
+        )
         user_service = pretend.stub(
             find_userid=pretend.call_recorder(lambda userid: 1),
             check_password=pretend.call_recorder(
@@ -150,7 +162,12 @@ def test_validate_password_ok(self):
         ]
 
     def test_validate_password_notok(self, db_session):
-        request = pretend.stub(remote_addr="127.0.0.1")
+        request = pretend.stub(
+            remote_addr="1.2.3.4",
+            banned=pretend.stub(
+                by_ip=lambda ip_address: False,
+            ),
+        )
         user_service = pretend.stub(
             find_userid=pretend.call_recorder(lambda userid: 1),
             check_password=pretend.call_recorder(
@@ -186,7 +203,12 @@ def test_validate_password_notok(self, db_session):
         ]
 
     def test_validate_password_too_many_failed(self):
-        request = pretend.stub(remote_addr="1.2.3.4")
+        request = pretend.stub(
+            remote_addr="1.2.3.4",
+            banned=pretend.stub(
+                by_ip=lambda ip_address: False,
+            ),
+        )
         user_service = pretend.stub(
             find_userid=pretend.call_recorder(lambda userid: 1),
             check_password=pretend.call_recorder(
@@ -218,7 +240,12 @@ def test_password_breached(self, monkeypatch):
         monkeypatch.setattr(forms, "send_password_compromised_email_hibp", send_email)
 
         user = pretend.stub(id=1)
-        request = pretend.stub(remote_addr="1.2.3.4")
+        request = pretend.stub(
+            remote_addr="1.2.3.4",
+            banned=pretend.stub(
+                by_ip=lambda ip_address: False,
+            ),
+        )
         user_service = pretend.stub(
             find_userid=lambda _: 1,
             get_user=lambda _: user,
@@ -247,6 +274,72 @@ def test_password_breached(self, monkeypatch):
         ]
         assert send_email.calls == [pretend.call(request, user)]
 
+    def test_validate_password_ok_ip_banned(self):
+        request = pretend.stub(
+            remote_addr="1.2.3.4",
+            banned=pretend.stub(
+                by_ip=lambda ip_address: True,
+            ),
+        )
+        user_service = pretend.stub(
+            find_userid=pretend.call_recorder(lambda userid: 1),
+            check_password=pretend.call_recorder(
+                lambda userid, password, tags=None: True
+            ),
+            is_disabled=pretend.call_recorder(lambda userid: (False, None)),
+        )
+        breach_service = pretend.stub(
+            check_password=pretend.call_recorder(lambda pw, tags: False)
+        )
+        form = forms.LoginForm(
+            data={"username": "my_username"},
+            request=request,
+            user_service=user_service,
+            breach_service=breach_service,
+            check_password_metrics_tags=["bar"],
+        )
+        field = pretend.stub(data="pw")
+
+        with pytest.raises(wtforms.validators.ValidationError):
+            form.validate_password(field)
+
+        assert user_service.find_userid.calls == []
+        assert user_service.is_disabled.calls == []
+        assert user_service.check_password.calls == []
+        assert breach_service.check_password.calls == []
+
+    def test_validate_password_notok_ip_banned(self, db_session):
+        request = pretend.stub(
+            remote_addr="1.2.3.4",
+            banned=pretend.stub(
+                by_ip=lambda ip_address: True,
+            ),
+        )
+        user_service = pretend.stub(
+            find_userid=pretend.call_recorder(lambda userid: 1),
+            check_password=pretend.call_recorder(
+                lambda userid, password, tags=None: False
+            ),
+            is_disabled=pretend.call_recorder(lambda userid: (False, None)),
+            record_event=pretend.call_recorder(lambda *a, **kw: None),
+        )
+        breach_service = pretend.stub()
+        form = forms.LoginForm(
+            data={"username": "my_username"},
+            request=request,
+            user_service=user_service,
+            breach_service=breach_service,
+        )
+        field = pretend.stub(data="pw")
+
+        with pytest.raises(wtforms.validators.ValidationError):
+            form.validate_password(field)
+
+        assert user_service.find_userid.calls == []
+        assert user_service.is_disabled.calls == []
+        assert user_service.check_password.calls == []
+        assert user_service.record_event.calls == []
+
 
 class TestRegistrationForm:
     def test_create(self):

tests/unit/accounts/test_models.py

@@ -103,15 +103,22 @@ def test_query_by_email_when_not_primary(self, db_session):
     def test_recent_events(self, db_session):
         user = DBUserFactory.create()
         recent_event = DBUserEventFactory(source=user, tag="foo", ip_address="0.0.0.0")
+        legacy_event = DBUserEventFactory(
+            source=user,
+            tag="wu",
+            ip_address_string="0.0.0.0",
+            time=datetime.datetime.now() - datetime.timedelta(days=1),
+        )
         stale_event = DBUserEventFactory(
             source=user,
             tag="bar",
             ip_address="0.0.0.0",
             time=datetime.datetime.now() - datetime.timedelta(days=91),
         )
 
-        assert user.events.all() == [recent_event, stale_event]
-        assert user.recent_events.all() == [recent_event]
+        assert user.events.all() == [recent_event, legacy_event, stale_event]
+        assert user.recent_events.all() == [recent_event, legacy_event]
+        assert user.recent_events.all()[-1].ip_address == "0.0.0.0"
 
     def test_regular_user_not_prohibited_password_reset(self, db_session):
         user = DBUserFactory.create()