fix(gitlab): sanitize ANSI escape sequences from MR webhook author email/name

src/sentry/integrations/gitlab/webhooks.py

@@ -2,6 +2,7 @@
 
 import inspect
 import logging
+import re
 from abc import ABC
 from collections.abc import Mapping
 from datetime import timezone
@@ -52,6 +53,45 @@
 logger = logging.getLogger("sentry.webhooks")
 
 PROVIDER_NAME = "integrations:gitlab"
+
+# Matches ANSI/VT100 CSI escape sequences such as \x1b[d or \x1b[0m, plus
+# any two-character escape sequence that starts with ESC but is not a CSI
+# introducer.  These sequences arrive in GitLab webhook payloads and consist
+# of a non-printable ESC byte followed by printable ASCII characters — so a
+# single-pass strip of non-printable bytes would leave the printable tail
+# ("[d", "[c", …) behind.
+_ANSI_ESCAPE_RE = re.compile(r"\x1b(?:\[[0-9;]*[A-Za-z]|[^\[])")
+
+# After stripping ANSI sequences, remove any remaining character outside the
+# printable ASCII range (0x20–0x7E), e.g. lone ESC bytes or other control
+# characters.
+_NON_PRINTABLE_ASCII_RE = re.compile(r"[^\x20-\x7E]")
+
+
+def _sanitize_author_field(value: str | None, max_length: int) -> str | None:
+    """Strip non-printable and ANSI escape characters, then truncate to *max_length*.
+
+    GitLab webhook payloads can carry ANSI escape sequences or other control
+    characters inside author email/name fields.  Passing those values straight
+    to ``CommitAuthor.objects.get_or_create`` causes PostgreSQL to raise a
+    ``StringDataRightTruncation`` (Django ``DataError``) when the value
+    exceeds the column's ``varchar(N)`` limit.  Sanitizing at the extraction
+    point — before the value touches any model layer — is the correct fix.
+
+    Two-pass approach:
+    1. Remove full ANSI/VT100 escape sequences (e.g. ``\\x1b[d``, ``\\x1b[0m``).
+       These sequences contain printable ASCII characters (``[``, ``d``, …) that
+       the second pass would not remove on its own.
+    2. Remove any remaining non-printable ASCII characters (control chars, lone
+       ESC bytes, non-ASCII code points).
+    """
+    if value is None:
+        return None
+    value = _ANSI_ESCAPE_RE.sub("", value)
+    value = _NON_PRINTABLE_ASCII_RE.sub("", value)
+    return value[:max_length]
+
+
 GITLAB_WEBHOOK_SECRET_INVALID_ERROR = """Gitlab's webhook secret does not match. Refresh token (or re-install the integration) by following this https://docs.sentry.io/organization/integrations/integration-platform/public-integration/#refreshing-tokens."""
 
 
@@ -456,8 +496,12 @@ def __call__(self, event: Mapping[str, Any], **kwargs):
             author_name = None
             head_commit_sha = None
             if last_commit:
-                author_email = last_commit["author"]["email"]
-                author_name = last_commit["author"]["name"]
+                author_email = _sanitize_author_field(
+                    last_commit["author"]["email"], max_length=200
+                )
+                author_name = _sanitize_author_field(
+                    last_commit["author"]["name"], max_length=128
+                )
                 head_commit_sha = last_commit.get("id")
 
             updated_at = event["object_attributes"].get("updated_at")
@@ -654,7 +698,8 @@ def __call__(self, event: Mapping[str, Any], **kwargs):
             if IntegrationRepositoryProvider.should_ignore_commit(commit["message"]):
                 continue
 
-            author_email = commit["author"]["email"]
+            author_email = _sanitize_author_field(commit["author"]["email"], max_length=200)
+            author_name = _sanitize_author_field(commit["author"]["name"], max_length=128)
 
             # TODO(dcramer): we need to deal with bad values here, but since
             # its optional, lets just throw it out for now
@@ -664,7 +709,7 @@ def __call__(self, event: Mapping[str, Any], **kwargs):
                 authors[author_email] = author = CommitAuthor.objects.get_or_create(
                     organization_id=organization.id,
                     email=author_email,
-                    defaults={"name": commit["author"]["name"]},
+                    defaults={"name": author_name},
                 )[0]
             else:
                 author = authors[author_email]

tests/sentry/integrations/gitlab/test_webhook.py

@@ -431,6 +431,42 @@ def test_merge_event_create_pull_request(self) -> None:
         self.assert_pull_request(pull, author)
         self.assert_group_link(group, pull)
 
+    def test_merge_event_author_email_with_ansi_sequences_is_sanitized(self) -> None:
+        # Regression test for SENTRY-5RKP: GitLab MR webhooks can carry ANSI
+        # escape sequences appended to the commit author email.  Passing the
+        # raw value to CommitAuthor.get_or_create caused PostgreSQL to raise a
+        # StringDataRightTruncation (DataError) when the value exceeded the
+        # email column's varchar(200) limit.  The fix strips all non-printable
+        # ASCII characters at the extraction point before any model layer is
+        # touched.
+        self.create_gitlab_repo("getsentry/sentry")
+        payload = orjson.loads(MERGE_REQUEST_OPENED_EVENT)
+
+        # Simulate the malformed email seen in production: a valid address
+        # followed by many ANSI cursor-movement escape sequences.
+        clean_email = "user@example.com"
+        ansi_suffix = "\x1b[d" * 34 + "\x1b[c" * 33  # 34 cursor-left + 33 cursor-right
+        dirty_email = clean_email + ansi_suffix
+        assert len(dirty_email) > 200  # would exceed varchar(200) without the fix
+
+        payload["object_attributes"]["last_commit"]["author"]["email"] = dirty_email
+        payload["object_attributes"]["last_commit"]["author"]["name"] = (
+            "Dev User\x1b[0m"  # also sanitize name
+        )
+
+        response = self.client.post(
+            self.url,
+            data=orjson.dumps(payload),
+            content_type="application/json",
+            HTTP_X_GITLAB_TOKEN=WEBHOOK_TOKEN,
+            HTTP_X_GITLAB_EVENT="Merge Request Hook",
+        )
+        assert response.status_code == 204
+
+        author = CommitAuthor.objects.get()
+        assert author.email == clean_email
+        assert author.name == "Dev User"
+
     def test_merge_event_update_pull_request(self) -> None:
         repo = self.create_gitlab_repo("getsentry/sentry")
         group = self.create_group(project=self.project, short_id=9)