Fix PostgreSQL binary-field corruption before encryption

CHANGELOG.md

@@ -13,6 +13,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Encrypt binary buffer contents before driver adaptation, fixing PostgreSQL writes that encrypted the string representation of a `psycopg.Binary` adapter. Previously corrupted values require application-specific recovery; this fix does not rewrite stored rows.
+
 - Synchronize primitive construction with cache invalidation so an in-flight load cannot republish a stale primitive after `clear_keyset_cache()`.
 
 - Restore the database timezone when decrypting naive datetime representations under `USE_TZ=True`, preserving instants across reads, re-saves, and deterministic lookups without rewriting stored ciphertext.

pyproject.toml

@@ -52,6 +52,7 @@ test = [
     "pytest>=8.3",
     "pytest-cov>=6",
     "pytest-django>=4.9",
+    "psycopg[binary]>=3.2",
 ]
 
 [tool.setuptools]

tink_fields/fields.py

@@ -625,6 +625,13 @@ class EncryptedBinaryField(EncryptedField, models.BinaryField):
     not be converted to strings during decryption.
     """
 
+    def _prepare_value_for_database(self, value: Any, connection: Any) -> bytes | None:
+        """Keep buffer contents as plaintext; adapt only the final ciphertext."""
+        value = self.get_prep_value(value)
+        if value is None:
+            return None
+        return bytes(memoryview(value))
+
     def _to_python_prepare(self, value: bytes) -> bytes:
         """Prepare decrypted value for to_python conversion.
 

tink_fields/test/test_binary_adaptation.py

@@ -0,0 +1,46 @@
+"""Use the real PostgreSQL driver adapter without needing a server."""
+
+from unittest.mock import patch
+
+import pytest
+from django.db import connection
+from django.db.backends.postgresql.base import DatabaseWrapper
+
+from tink_fields import EncryptedBinaryField
+
+from . import models
+
+
+@pytest.mark.parametrize("value", [b"", b"binary\x00\xff", bytearray(b"mutable\xff"), memoryview(b"view\x00")])
+def test_postgresql_binary_adapter_wraps_ciphertext_only(value):
+    postgres = DatabaseWrapper({})
+    field = EncryptedBinaryField()
+    with patch.object(postgres.Database, "Binary", wraps=postgres.Database.Binary) as binary:
+        adapted = field.get_db_prep_save(value, postgres)
+    assert field.from_db_value(adapted.obj, None, postgres) == bytes(value)
+    assert binary.call_count == 1
+    assert bytes(adapted.obj) != bytes(value)
+
+
+def test_postgresql_binary_null_skips_adaptation():
+    postgres = DatabaseWrapper({})
+    with patch.object(postgres.Database, "Binary", wraps=postgres.Database.Binary) as binary:
+        assert EncryptedBinaryField().get_db_prep_save(None, postgres) is None
+    binary.assert_not_called()
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize("value", [b"", b"binary\x00\xff", bytearray(b"mutable\xff"), memoryview(b"view\x00")])
+def test_binary_buffer_values_round_trip_in_database(value):
+    instance = models.EncryptedBinary.objects.create(value=value)
+    instance.refresh_from_db()
+    assert instance.value == bytes(value)
+    with connection.cursor() as cursor:
+        cursor.execute(f"SELECT value FROM {models.EncryptedBinary._meta.db_table} WHERE id = %s", [instance.pk])
+        assert bytes(cursor.fetchone()[0]) != bytes(value)
+
+
+@pytest.mark.parametrize("value", ["not binary", 3])
+def test_non_buffer_values_are_rejected(value):
+    with pytest.raises(TypeError):
+        EncryptedBinaryField().get_db_prep_save(value, connection)