diff --git a/CHANGELOG.md b/CHANGELOG.md index d12e8af..8cc35bd 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 315ae2c..4702815 100644 --- a/README.md +++ b/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 diff --git a/tink_fields/fields.py b/tink_fields/fields.py index 8d58c81..d1cf9e7 100644 --- a/tink_fields/fields.py +++ b/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() diff --git a/tink_fields/test/test_lookup_guards.py b/tink_fields/test/test_lookup_guards.py new file mode 100644 index 0000000..442f22d --- /dev/null +++ b/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