fix(replay): validate IPv4 queries (#69326)

Fixes [SNUBA-4P4](https://sentry.sentry.io/issues/5221030000/)
Reproduce:
https://sentry.sentry.io/replays/?project=11276&query=user.ip%3A%5B192%2C+0%2C+0%2C+1%5D&statsPeriod=14d

UI will show the same error, but we'll avoid the SnubaError

src/sentry/replays/lib/new_query/parsers.py

@@ -2,7 +2,7 @@
 
 Functions in this module coerce external types to internal types.  Else they die.
 """
-
+import ipaddress
 import uuid
 
 from sentry.replays.lib.new_query.errors import CouldNotParseValue
@@ -26,6 +26,15 @@ def parse_str(value: str) -> str:
     return value
 
 
+def parse_ipv4(value: str) -> str:
+    """Validates an IPv4 address"""
+    try:
+        ipaddress.IPv4Address(value)
+        return value
+    except ipaddress.AddressValueError:
+        raise CouldNotParseValue("Invalid IPv4")
+
+
 def parse_uuid(value: str) -> uuid.UUID:
     try:
         return uuid.UUID(value)

src/sentry/replays/usecases/query/configs/aggregate.py

@@ -24,7 +24,7 @@ from sentry.replays.lib.new_query.fields import (
     SumLengthField,
     UUIDColumnField,
 )
-from sentry.replays.lib.new_query.parsers import parse_int, parse_str, parse_uuid
+from sentry.replays.lib.new_query.parsers import parse_int, parse_ipv4, parse_str, parse_uuid
 from sentry.replays.lib.selector.parse import parse_selector
 from sentry.replays.usecases.query.conditions import (
     AggregateActivityScalar,
@@ -120,7 +120,7 @@ search_config: dict[str, FieldProtocol] = {
     "urls": array_string_field("urls"),
     "user.email": string_field("user_email"),
     "user.id": string_field("user_id"),
-    "user.ip_address": StringColumnField("ip_address_v4", parse_str, SumOfIPv4Scalar),
+    "user.ip_address": StringColumnField("ip_address_v4", parse_ipv4, SumOfIPv4Scalar),
     "user.username": string_field("user_name"),
     "viewed_by_id": IntegerColumnField("viewed_by_id", parse_int, SumOfIntegerIdScalar),
     "warning_ids": UUIDColumnField("warning_id", parse_uuid, SumOfUUIDScalar),

src/sentry/replays/usecases/query/configs/scalar.py

@@ -12,7 +12,7 @@ from sentry.replays.lib.new_query.conditions import (
     UUIDArray,
 )
 from sentry.replays.lib.new_query.fields import FieldProtocol, StringColumnField, UUIDColumnField
-from sentry.replays.lib.new_query.parsers import parse_str, parse_uuid
+from sentry.replays.lib.new_query.parsers import parse_ipv4, parse_str, parse_uuid
 from sentry.replays.lib.selector.parse import parse_selector
 from sentry.replays.usecases.query.conditions import (
     ClickSelectorComposite,
@@ -61,7 +61,7 @@ varying_search_config: dict[str, FieldProtocol] = {
     "urls": StringColumnField("urls", parse_str, StringArray),
     "user.email": StringColumnField("user_email", parse_str, NonEmptyStringScalar),
     "user.id": StringColumnField("user_id", parse_str, NonEmptyStringScalar),
-    "user.ip_address": StringColumnField("ip_address_v4", parse_str, IPv4Scalar),
+    "user.ip_address": StringColumnField("ip_address_v4", parse_ipv4, IPv4Scalar),
     "user.username": StringColumnField("user_name", parse_str, NonEmptyStringScalar),
 }
 

tests/sentry/replays/test_organization_replay_index.py

@@ -1720,6 +1720,27 @@ class OrganizationReplayIndexTest(APITestCase, ReplaysSnubaTestCase):
                 response_data = response.json()
                 assert len(response_data["data"]) == 1, query
 
+    def test_query_invalid_ipv4_addresses(self):
+        project = self.create_project(teams=[self.team])
+
+        replay1_id = uuid.uuid4().hex
+        seq1_timestamp = datetime.datetime.now() - datetime.timedelta(seconds=22)
+        seq2_timestamp = datetime.datetime.now() - datetime.timedelta(seconds=5)
+
+        self.store_replays(mock_replay(seq1_timestamp, project.id, replay1_id))
+        self.store_replays(mock_replay(seq2_timestamp, project.id, replay1_id))
+
+        with self.feature(REPLAYS_FEATURES):
+            queries = [
+                "user.ip:127.256.0.1",
+                "!user.ip_address:192.168.z34.1",
+                "user.ip_address:bacontest",
+                "user.ip_address:[127.0.0.,192.168.0.1]",
+            ]
+            for query in queries:
+                response = self.client.get(self.url + f"?field=id&query={query}")
+                assert response.status_code == 400
+
     def test_query_branches_computed_activity_conditions(self):
         project = self.create_project(teams=[self.team])