Fix plaintext lookup and transform bypasses on encrypted fields

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Reject inherited JSON/date transforms and late-registered plaintext lookups on encrypted columns; keep deterministic exact and SQL null lookups explicit.
+
 ## [0.4.0] - 2026-08-01
 
 ### Added

README.md

@@ -93,7 +93,7 @@ Values are ordinary Python objects on model instances. Django validates them usi
 | `EncryptedURLField` | `URLField` |
 | `EncryptedUUIDField` | `UUIDField` |
 
-Randomized fields deliberately reject `primary_key`, `unique`, `db_index`, and `db_default`. They support `isnull` queries, including the equivalent `field=None`; every lookup that compares values raises `FieldError`. Database expressions such as `F()` assignments are also rejected because the database cannot encrypt them.
+Randomized fields deliberately reject `primary_key`, `unique`, `db_index`, and `db_default`. They support `isnull` queries, including the equivalent `field=None`; every lookup that compares values raises `FieldError`. Inherited JSON key lookups, date transforms, and custom registered lookups are also rejected because they would operate on ciphertext. Database expressions such as `F()` assignments are also rejected because the database cannot encrypt them.
 
 ### Deterministic fields and exact lookups
 

tink_fields/fields.py

@@ -20,6 +20,7 @@
 from django.conf import settings
 from django.core.exceptions import FieldError, ImproperlyConfigured
 from django.db import models
+from django.db.models.lookups import Exact, IsNull, Lookup
 from django.utils.encoding import force_bytes, force_str
 from django.utils.functional import cached_property
 from tink import JsonKeysetReader, TinkError, aead, cleartext_keyset_handle, daead, read_keyset_handle
@@ -372,6 +373,18 @@ def get_internal_type(self) -> str:
         """
         return self._internal_type
 
+    def get_lookup(self, lookup_name: str) -> type[Lookup]:
+        """Select only operations that are meaningful for ciphertext."""
+        if lookup_name == "isnull":
+            return IsNull
+        if lookup_name == "exact":
+            return EncryptedExact
+        raise FieldError(f"{self.__class__.__name__} `{lookup_name}` does not support lookups.")
+
+    def get_transform(self, lookup_name: str) -> None:
+        """Prevent inherited date and JSON transforms from inspecting ciphertext."""
+        raise FieldError(f"{self.__class__.__name__} `{lookup_name}` does not support lookups.")
+
     def get_db_prep_save(self, value: Any, connection: Any) -> Any:
         """Prepare the value for saving to the database.
 
@@ -447,55 +460,26 @@ def __repr__(self) -> str:
         return f"<{self.__class__.__name__}: keyset={self._keyset}>"
 
 
-def _create_lookup_class(lookup_name: str, base_lookup_class: type[Any]) -> type[Any]:
-    """Create a lookup class that raises errors for encrypted fields.
-
-    Args:
-        lookup_name: Name of the lookup operation
-        base_lookup_class: Base lookup class to inherit from
-
-    Returns:
-        type: New lookup class that raises FieldError
-    """
+class EncryptedExact(Exact):
+    """Allow Django to rewrite equality with None to an IS NULL lookup."""
 
     def get_prep_lookup(self) -> Any:
-        """Raise error for unsupported lookups."""
-        if self.lookup_name == "exact" and self.rhs is None:
+        if self.rhs is None:
             return None
-        raise FieldError(f"{self.lhs.field.__class__.__name__} `{self.lookup_name}` does not support lookups.")
+        field = self.lhs.output_field
+        raise FieldError(f"{field.__class__.__name__} `exact` does not support lookups.")
 
-    return type(
-        f"EncryptedField{lookup_name}",
-        (base_lookup_class,),
-        {"get_prep_lookup": get_prep_lookup},
-    )
 
-
-def _create_deterministic_lookup_class(lookup_name: str, base_lookup_class: type[Any]) -> type[Any]:
-    """Create a lookup class for deterministic encrypted fields.
-
-    For deterministic fields, we support exact lookups by encrypting the
-    prepared value and comparing ciphertexts.
-
-    Args:
-        lookup_name: Name of the lookup operation
-        base_lookup_class: Base lookup class to inherit from
-
-    Returns:
-        type: New lookup class for deterministic fields
-    """
+class DeterministicEncryptedExact(Exact):
+    """Compare ciphertext prepared using the same conversion as writes."""
 
     def get_prep_lookup(self) -> Any:
-        """Handle lookups for deterministic encrypted fields."""
-        if self.lookup_name == "exact":
-            if hasattr(self.rhs, "resolve_expression"):
-                raise FieldError("Deterministic encrypted lookups do not support database expressions.")
-            return self.rhs
-        raise FieldError(f"{self.lhs.field.__class__.__name__} `{self.lookup_name}` does not support lookups.")
+        if hasattr(self.rhs, "resolve_expression"):
+            raise FieldError("Deterministic encrypted lookups do not support database expressions.")
+        return self.rhs
 
     def get_db_prep_lookup(self, value: Any, connection: Any) -> tuple[str, list[Any]]:
-        """Prepare and deterministically encrypt an exact lookup value."""
-        field = self.lhs.field
+        field = self.lhs.output_field
         prepared_value = field._prepare_value_for_database(value, connection)
         if prepared_value is None:
             return "%s", [None]
@@ -511,29 +495,6 @@ def as_sql(self, compiler: Any, connection: Any) -> tuple[str, tuple[Any, ...]]:
         rhs_sql = self.get_rhs_op(connection, rhs_sql)
         return f"{lhs_sql} {rhs_sql}", (*lhs_params, *rhs_params)
 
-    return type(
-        f"DeterministicEncryptedField{lookup_name}",
-        (base_lookup_class,),
-        {
-            "get_prep_lookup": get_prep_lookup,
-            "get_db_prep_lookup": get_db_prep_lookup,
-            "as_sql": as_sql,
-        },
-    )
-
-
-def _register_lookup_classes():
-    """Register lookup classes for encrypted fields."""
-    for name, lookup in models.Field.class_lookups.items():
-        if name != "isnull":
-            # Register lookup class for regular encrypted fields
-            lookup_class = _create_lookup_class(name, lookup)
-            EncryptedField.register_lookup(lookup_class)
-
-            # Register lookup class for deterministic encrypted fields
-            deterministic_lookup_class = _create_deterministic_lookup_class(name, lookup)
-            DeterministicEncryptedField.register_lookup(deterministic_lookup_class)
-
 
 # Field implementations
 class EncryptedTextField(EncryptedField, models.TextField):
@@ -671,6 +632,11 @@ class DeterministicEncryptedField(EncryptedField):
 
     _unsupported_properties = frozenset(["primary_key", "db_default"])
 
+    def get_lookup(self, lookup_name: str) -> type[Lookup]:
+        if lookup_name == "exact":
+            return DeterministicEncryptedExact
+        return super().get_lookup(lookup_name)
+
     def get_db_prep_save(self, value: Any, connection: Any) -> Any:
         """Prepare the value for saving to the database using deterministic encryption.
 
@@ -761,7 +727,3 @@ class DeterministicEncryptedDateTimeField(DeterministicEncryptedField, models.Da
     """Deterministic encrypted datetime field."""
 
     pass
-
-
-# Register lookup classes at module level
-_register_lookup_classes()

tink_fields/test/test_lookup_guards.py

@@ -0,0 +1,57 @@
+"""Encrypted columns must never inherit plaintext lookups or transforms."""
+
+import pytest
+from django.core.exceptions import FieldError
+from django.db import connection
+from django.db.models import Field, Lookup
+from django.test.utils import register_lookup
+
+from . import models
+
+
+@pytest.mark.parametrize(
+    "model, lookup, value",
+    [
+        (models.EncryptedExtended, "payload__has_key", "secret"),
+        (models.EncryptedExtended, "payload__has_keys", ["secret"]),
+        (models.EncryptedExtended, "payload__secret", "value"),
+        (models.EncryptedExtended, "payload__secret__isnull", True),
+        (models.EncryptedDate, "value__year", 2026),
+        (models.EncryptedDateTime, "value__date", "2026-09-05"),
+        (models.DeterministicEncryptedExtended, "day__year", 2026),
+        (models.DeterministicEncryptedExtended, "moment__date", "2026-09-05"),
+    ],
+)
+def test_plaintext_lookups_and_transforms_are_rejected(model, lookup, value):
+    with pytest.raises(FieldError, match="does not support lookups"):
+        model.objects.filter(**{lookup: value})
+
+
+@pytest.mark.parametrize("lookup", ["payload__secret", "payload__secret__nested"])
+def test_json_projection_transforms_are_rejected(lookup):
+    with pytest.raises(FieldError, match="Cannot resolve keyword"):
+        models.EncryptedExtended.objects.values(lookup)
+
+
+@pytest.mark.parametrize("model", [models.EncryptedText, models.DeterministicEncryptedText])
+def test_late_registered_lookup_cannot_bypass_encryption_guards(model):
+    class PlaintextLookup(Lookup):
+        lookup_name = "plaintext_probe"
+
+        def as_sql(self, compiler, connection):
+            return "1 = 1", []
+
+    with register_lookup(Field, PlaintextLookup), pytest.raises(FieldError, match="does not support lookups"):
+        model.objects.filter(value__plaintext_probe="secret")
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize("model", [models.EncryptedNullable, models.DeterministicEncryptedTextNullable])
+def test_null_lookups_keep_sql_null_semantics(model):
+    instance = model.objects.create(value=None)
+    assert model.objects.get(value=None).pk == instance.pk
+    assert model.objects.get(value__isnull=True).pk == instance.pk
+    assert not model.objects.filter(value__isnull=False).exists()
+    with connection.cursor() as cursor:
+        cursor.execute(f"SELECT value FROM {model._meta.db_table} WHERE id = %s", [instance.pk])
+        assert cursor.fetchone()[0] is None