Also sanitize edge identifiers

Author: emyller

api/edge_api/identities/serializers.py

@@ -18,6 +18,7 @@
 from rest_framework.exceptions import ValidationError
 
 from environments.dynamodb.types import IdentityOverrideV2
+from environments.identities.constants import identifier_regex_validator
 from environments.models import Environment
 from features.models import Feature, FeatureState, FeatureStateValue
 from features.multivariate.models import MultivariateFeatureOption
@@ -52,7 +53,11 @@ def to_internal_value(self, data: typing.Any) -> str:
 
 class EdgeIdentitySerializer(serializers.Serializer):  # type: ignore[type-arg]
     identity_uuid = serializers.CharField(read_only=True)
-    identifier = serializers.CharField(required=True, max_length=2000)
+    identifier = serializers.CharField(
+        required=True,
+        max_length=2000,
+        validators=[identifier_regex_validator],
+    )
     dashboard_alias = LowerCaseCharField(
         required=False,
         max_length=100,

api/environments/identities/constants.py

@@ -0,0 +1,10 @@
+import re
+
+from django.core.validators import RegexValidator
+
+RE_VALID_IDENTIFIER = re.compile(r"^[\w!#$%&*+/=?^_`{}|~@.\-]+$")
+
+identifier_regex_validator = RegexValidator(
+    regex=RE_VALID_IDENTIFIER,
+    message="Identifier can only contain unicode letters, numbers, and the symbols: ! # $ %% & * + / = ? ^ _ ` { } | ~ @ . -",
+)

api/environments/identities/migrations/0003_sanitized_identifiers.py

@@ -1,5 +1,7 @@
 # Generated by Django 4.2.22 on 2025-09-08 23:30
 
+import re
+
 import django.core.validators
 from django.db import migrations, models
 
@@ -19,7 +21,7 @@ class Migration(migrations.Migration):
                 validators=[
                     django.core.validators.RegexValidator(
                         message="Identifier can only contain unicode letters, numbers, and the symbols: ! # $ %% & * + / = ? ^ _ ` { } | ~ @ . -",
-                        regex="^[\\w!#$%&*+/=?^_`{}|~@.\\-]+$",
+                        regex=re.compile("^[\\w!#$%&*+/=?^_`{}|~@.\\-]+$"),
                     )
                 ],
             ),

api/environments/identities/models.py

@@ -1,11 +1,11 @@
 from itertools import chain
 
-from django.core.validators import RegexValidator
 from django.db import models
 from django.db.models import Prefetch, Q
 from flag_engine.context.mappers import map_environment_identity_to_context
 from flag_engine.segments.evaluator import is_context_in_segment
 
+from environments.identities.constants import identifier_regex_validator
 from environments.identities.managers import IdentityManager
 from environments.identities.traits.models import Trait
 from environments.models import Environment
@@ -24,12 +24,7 @@
 class Identity(models.Model):
     identifier = models.CharField(
         max_length=2000,
-        validators=[
-            RegexValidator(
-                regex=r"^[\w!#$%&*+/=?^_`{}|~@.\-]+$",
-                message="Identifier can only contain unicode letters, numbers, and the symbols: ! # $ %% & * + / = ? ^ _ ` { } | ~ @ . -",
-            ),
-        ],
+        validators=[identifier_regex_validator],
     )
     created_date = models.DateTimeField("DateCreated", auto_now_add=True)
     environment = models.ForeignKey(

api/tests/integration/edge_api/identities/test_edge_identity_viewset.py

@@ -1,7 +1,9 @@
 import json
 import urllib
 from typing import Any
+from unittest.mock import Mock
 
+import pytest
 from boto3.dynamodb.conditions import Key
 from django.urls import reverse
 from mypy_boto3_dynamodb.service_resource import Table
@@ -18,6 +20,8 @@
 )
 from environments.models import Environment
 
+_invalid_identifier_error_message = "Identifier can only contain unicode letters, numbers, and the symbols: ! # $ % & * + / = ? ^ _ ` { } | ~ @ . -"
+
 
 def test_get_identities_returns_bad_request_if_dynamo_is_not_enabled(  # type: ignore[no-untyped-def]
     admin_client, environment, environment_api_key
@@ -116,6 +120,78 @@ def test_create_identity(  # type: ignore[no-untyped-def]
     assert response.json()["identity_uuid"] is not None
 
 
+@pytest.mark.parametrize(
+    "given_identifier",
+    [
+        "bond...jamesbond",
+        "ゴジラ",
+        "ElChapulínColorado",
+        "dalek#[email protected]",
+        "agáta={^_^}=",
+        "_ツ_/-handless-shrug",
+        "who+am+i?",
+        "i_100%_dont_know!",
+        "~neo|simulation`0065192*75`",
+        "KacperGustyr$Flagsmat",
+    ],
+)
+@pytest.mark.usefixtures(
+    "dynamo_enabled_environment",
+)
+def test_create_identity_accepts_valid_identifiers(
+    admin_client: APIClient,
+    environment_api_key: str,
+    given_identifier: str,
+    edge_identity_dynamo_wrapper_mock: Mock,
+) -> None:
+    # Given
+    edge_identity_dynamo_wrapper_mock.get_item.return_value = None
+
+    # When
+    response = admin_client.post(
+        f"/api/v1/environments/{environment_api_key}/edge-identities/",
+        data={"identifier": given_identifier},
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_201_CREATED
+    assert response.json()["identifier"] == given_identifier
+
+
+@pytest.mark.parametrize(
+    ["given_identifier", "error_message"],
+    [
+        ("", "This field may not be blank."),
+        (" ", "This field may not be blank."),
+        ("or really anything with a whitespace", _invalid_identifier_error_message),
+        ("<script>alert(1)</script>", _invalid_identifier_error_message),
+        ("'; DROP TABLE users;--", _invalid_identifier_error_message),
+        ("'single-quotes'", _invalid_identifier_error_message),
+        ('"double-quotes"', _invalid_identifier_error_message),
+        ("figaro" * 334, "Ensure this field has no more than 2000 characters."),
+    ],
+)
+@pytest.mark.usefixtures(
+    "dynamo_enabled_environment",
+    "edge_identity_dynamo_wrapper_mock",
+)
+def test_create_identity_responds_400_if_identifier_is_invalid(
+    admin_client: APIClient,
+    environment_api_key: str,
+    error_message: str,
+    given_identifier: str,
+) -> None:
+    # When
+    response = admin_client.post(
+        f"/api/v1/environments/{environment_api_key}/edge-identities/",
+        data={"identifier": given_identifier},
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.json() == {"identifier": [error_message]}
+
+
 def test_create_identity_returns_400_if_identity_already_exists(  # type: ignore[no-untyped-def]
     admin_client,
     dynamo_enabled_environment,