fix(crons): Properly checks for environment name lengths (#52820)

Safety check for environment name lengths

src/sentry/monitors/consumers/monitor_consumer.py

@@ -24,6 +24,7 @@
     MonitorCheckIn,
     MonitorEnvironment,
     MonitorEnvironmentLimitsExceeded,
+    MonitorEnvironmentValidationFailed,
     MonitorLimitsExceeded,
     MonitorType,
 )
@@ -283,6 +284,13 @@ def update_existing_check_in(
                 )
                 logger.debug("monitor environment exceeds limits for monitor: %s", monitor_slug)
                 return
+            except MonitorEnvironmentValidationFailed:
+                metrics.incr(
+                    "monitors.checkin.result",
+                    tags={**metric_kwargs, "status": "failed_monitor_environment_name_length"},
+                )
+                logger.debug("monitor environment name too long: %s %s", monitor_slug, environment)
+                return
 
             status = getattr(CheckInStatus, validated_params["status"].upper())
 

src/sentry/monitors/endpoints/monitor_ingest_checkin_index.py

@@ -28,6 +28,7 @@
     MonitorCheckIn,
     MonitorEnvironment,
     MonitorEnvironmentLimitsExceeded,
+    MonitorEnvironmentValidationFailed,
     MonitorLimitsExceeded,
 )
 from sentry.monitors.serializers import MonitorCheckInSerializerResponse
@@ -185,7 +186,7 @@ def post(
                 monitor_environment = MonitorEnvironment.objects.ensure_environment(
                     project, monitor, result.get("environment")
                 )
-            except MonitorEnvironmentLimitsExceeded as e:
+            except (MonitorEnvironmentLimitsExceeded, MonitorEnvironmentValidationFailed) as e:
                 return self.respond({type(e).__name__: str(e)}, status=403)
 
             # Infer the original start time of the check-in from the duration.

src/sentry/monitors/models.py

@@ -79,6 +79,10 @@ class MonitorEnvironmentLimitsExceeded(Exception):
     pass
 
 
+class MonitorEnvironmentValidationFailed(Exception):
+    pass
+
+
 def get_next_schedule(last_checkin, schedule_type, schedule):
     if schedule_type == ScheduleType.CRONTAB:
         itr = croniter(schedule, last_checkin)
@@ -447,6 +451,9 @@ def ensure_environment(
         if not environment_name:
             environment_name = "production"
 
+        if not Environment.is_valid_name(environment_name):
+            raise MonitorEnvironmentValidationFailed("Environment name too long")
+
         # TODO: assume these objects exist once backfill is completed
         environment = Environment.get_or_create(project=project, name=environment_name)
 

tests/sentry/monitors/endpoints/test_monitor_ingest_checkin_index.py

@@ -359,6 +359,25 @@ def test_monitor_environment_creation_over_limit(self):
             assert resp.status_code == 403
             assert "MonitorEnvironmentLimitsExceeded" in resp.data.keys()
 
+    def test_monitor_environment_validation(self):
+        for i, path_func in enumerate(self._get_path_functions()):
+            slug = f"my-new-monitor-{i}"
+            path = path_func(slug)
+
+            invalid_name = "x" * 65
+
+            resp = self.client.post(
+                path,
+                {
+                    "status": "ok",
+                    "monitor_config": {"schedule_type": "crontab", "schedule": "5 * * * *"},
+                    "environment": f"environment-{invalid_name}",
+                },
+                **self.dsn_auth_headers,
+            )
+            assert resp.status_code == 403
+            assert "MonitorEnvironmentValidationFailed" in resp.data.keys()
+
     def test_with_dsn_auth_and_guid(self):
         for path_func in self._get_path_functions():
             monitor = self._create_monitor()

tests/sentry/monitors/test_monitor_consumer.py

@@ -468,6 +468,21 @@ def test_monitor_environment_limits(self):
         monitor_environments = MonitorEnvironment.objects.filter(monitor=monitor)
         assert len(monitor_environments) == settings.MAX_ENVIRONMENTS_PER_MONITOR
 
+    def test_monitor_environment_validation(self):
+        invalid_name = "x" * 65
+
+        self.send_message(
+            "my-monitor",
+            monitor_config={"schedule": {"type": "crontab", "value": "13 * * * *"}},
+            environment=f"my-environment-{invalid_name}",
+        )
+
+        monitor = Monitor.objects.get(slug="my-monitor")
+        assert monitor is not None
+
+        monitor_environments = MonitorEnvironment.objects.filter(monitor=monitor)
+        assert len(monitor_environments) == 0
+
     def test_organization_killswitch(self):
         monitor = self._create_monitor(slug="my-monitor")