feat: Workspace 1074

Pipfile

@@ -22,7 +22,7 @@ regex = "==2024.11.6"
 requests = "==2.32.5"
 gunicorn = "==23.0.0"
 uvicorn-worker = "==0.2.0"
-pyjwt = "==2.6.0" # TODO: upgrade to latest version.
+pyjwt = "==2.12.1" 
 psutil = "==7.0.0"
 google-auth = "==2.48.0"
 google-cloud-bigquery = "==3.38.0"

codeforlife/__init__.py

@@ -6,7 +6,6 @@
 import os
 import sys
 import typing as t
-from io import StringIO
 from pathlib import Path
 from types import SimpleNamespace
 
@@ -113,20 +112,7 @@ def set_up_settings(service_base_dir: Path, service_name: str):
 
         secrets = dotenv_values(secrets_path)
     else:
-        # pylint: disable-next=import-outside-toplevel
-        import boto3
-
-        s3: "S3Client" = boto3.client("s3")
-        secrets_object = s3.get_object(
-            Bucket=os.environ["aws_s3_app_bucket"],
-            Key=(
-                os.environ["aws_s3_app_folder"]
-                + f"/secure/.env.secrets.{service_name}"
-            ),
-        )
-
-        secrets = dotenv_values(
-            stream=StringIO(secrets_object["Body"].read().decode("utf-8"))
-        )
+        # TODO: load secrets from bucket in non-local environments.
+        secrets = {}
 
     return Secrets(**secrets)

codeforlife/encryption.py

@@ -52,20 +52,22 @@
 class FakeAead:
     """A fake AEAD primitive for local testing."""
 
-    @staticmethod
+    ciphertext_prefix = b"fake_enc:"
+
+    @classmethod
     # pylint: disable-next=unused-argument
-    def encrypt(plaintext: bytes, associated_data: bytes = b""):
+    def encrypt(cls, plaintext: bytes, associated_data: bytes = b""):
         """Simulate ciphertext by wrapping in base64 and adding a prefix."""
-        return b"fake_enc:" + b64encode(plaintext)
+        return cls.ciphertext_prefix + b64encode(plaintext)
 
-    @staticmethod
+    @classmethod
     # pylint: disable-next=unused-argument
-    def decrypt(ciphertext: bytes, associated_data: bytes = b""):
+    def decrypt(cls, ciphertext: bytes, associated_data: bytes = b""):
         """Simulate decryption by removing prefix and base64 decoding."""
-        if not ciphertext.startswith(b"fake_enc:"):
+        if not ciphertext.startswith(cls.ciphertext_prefix):
             raise ValueError("Invalid ciphertext for fake mock")
 
-        return b64decode(ciphertext.replace(b"fake_enc:", b""))
+        return b64decode(ciphertext.replace(cls.ciphertext_prefix, b""))
 
     @classmethod
     def as_mock(cls):

codeforlife/models/encrypted.py

@@ -110,7 +110,8 @@ def _check_associated_data(cls, **kwargs):
         """
         errors: t.List[checks.Error] = []
 
-        if cls._meta.abstract:
+        # Skip abstract and proxy models.
+        if cls._meta.abstract or cls._meta.proxy:
             return errors
 
         # Ensure associated_data is defined.
@@ -150,6 +151,7 @@ def _check_associated_data(cls, **kwargs):
                     # pylint: disable-next=too-many-boolean-expressions
                     not model is cls
                     and not model._meta.abstract
+                    and not model._meta.proxy
                     and issubclass(model, EncryptedModel)
                     and hasattr(model, "associated_data")
                     and isinstance(model.associated_data, str)

codeforlife/models/fields/__init__.py

@@ -7,3 +7,4 @@
 from .data_encryption_key import DataEncryptionKeyField
 from .deferred_attribute import DeferredAttribute
 from .encrypted_text import EncryptedTextField
+from .sha256 import Sha256Field

codeforlife/models/fields/base_encrypted.py

@@ -51,15 +51,18 @@
 
 import typing as t
 from dataclasses import dataclass
+from enum import IntEnum, auto
 
 from django.core.exceptions import ValidationError
 from django.db.models import BinaryField
 
 from ...types import Args, KwArgs
 from ..encrypted import EncryptedModel
+from ..utils import is_real_model_class
 from .deferred_attribute import DeferredAttribute
 
 T = t.TypeVar("T")
+Ciphertext: t.TypeAlias = t.Union[bytes, memoryview]
 
 
 @dataclass(frozen=True)
@@ -73,7 +76,14 @@ class _PendingEncryption(t.Generic[T]):
 class _TrustedCiphertext:
     """A wrapper for ciphertext that comes directly from the database."""
 
-    ciphertext: bytes
+    class Source(IntEnum):
+        """The source of the ciphertext."""
+
+        DB = auto()
+        FIXTURE = auto()
+
+    ciphertext: Ciphertext
+    source: Source
 
 
 Value: t.TypeAlias = t.Union[_TrustedCiphertext, _PendingEncryption[T]]
@@ -108,15 +118,27 @@ def __get__(self, instance, cls=None):
             return internal_value.value
 
         if isinstance(internal_value, _TrustedCiphertext):
+            # If the ciphertext came from a fixture, do not decrypt it so that
+            # it can be loaded as-is into the database.
+            if internal_value.source == _TrustedCiphertext.Source.FIXTURE:
+                return internal_value.ciphertext
+
             # If we have a cached decrypted value, return it.
             if self.field.attname in instance.__decrypted_values__:
                 return t.cast(
                     T, instance.__decrypted_values__[self.field.attname]
                 )
 
+            # Extract the raw bytes from the ciphertext.
+            ciphertext = (
+                internal_value.ciphertext
+                if isinstance(internal_value.ciphertext, bytes)
+                else bytes(internal_value.ciphertext)
+            )
+
             # Decrypt the value before returning it.
             decrypted_value = t.cast(
-                T, self.field.decrypt_value(instance, internal_value.ciphertext)
+                T, self.field.decrypt_value(instance, ciphertext)
             )
 
             # Cache the decrypted value on the instance.
@@ -152,7 +174,9 @@ def __set__(
                     "Expected bytes in memoryview for encrypted field.",
                     code="invalid_memoryview_type",
                 )
-            internal_value = _TrustedCiphertext(value.obj)
+            internal_value = _TrustedCiphertext(
+                value, _TrustedCiphertext.Source.FIXTURE
+            )
         elif isinstance(value, _TrustedCiphertext):  # From DB.
             internal_value = value
         else:  # From user input.
@@ -173,33 +197,21 @@ class BaseEncryptedField(BinaryField, t.Generic[T]):
     # Construction & Deconstruction
     # --------------------------------------------------------------------------
 
-    def set_init_kwargs(self, kwargs: KwArgs):
-        """Sets common init kwargs."""
-        kwargs.setdefault("db_column", self.associated_data)
-
-    def __init__(
-        self,
-        associated_data: str,
-        # Set type for default to match T.
-        default: t.Optional[t.Union[T, t.Callable[[], T]]] = None,
-        **kwargs,
-    ):
+    def __init__(self, associated_data: str, **kwargs):
         if not associated_data:
             raise ValidationError(
                 "Associated data cannot be empty.",
                 code="no_associated_data",
             )
         self.associated_data = associated_data
 
-        self.set_init_kwargs(kwargs)
-        super().__init__(**kwargs, default=default)
+        super().__init__(**kwargs)
 
     def deconstruct(self):
         name, path, args, kwargs = t.cast(
             t.Tuple[str, str, Args, KwArgs], super().deconstruct()
         )
 
-        self.set_init_kwargs(kwargs)
         kwargs["associated_data"] = self.associated_data
 
         return name, path, args, kwargs
@@ -215,8 +227,8 @@ def contribute_to_class(self, cls, name, private_only=False):
         """
         super().contribute_to_class(cls, name, private_only)
 
-        # Skip fake models used for migrations.
-        if cls.__module__ == "__fake__":
+        # Skip fake (used for migrations), abstract and proxy models.
+        if not is_real_model_class(cls):
             return
 
         # Ensure the model subclasses EncryptedModel.
@@ -271,8 +283,7 @@ def __get__(self, instance: t.Optional[EncryptedModel], owner: t.Any):
     # Set the internal value when assigned on an instance.
     def __set__(self, instance: EncryptedModel, value: t.Optional[T]): ...
 
-    # pylint: disable-next=unused-argument
-    def from_db_value(self, value: t.Optional[bytes], expression, connection):
+    def from_db_value(self, value: t.Optional[Ciphertext], _, __):
         """
         Converts a value as returned by the database to a Python object.
         We wrap the raw bytes in _TrustedCiphertext to signal that this is
@@ -282,7 +293,7 @@ def from_db_value(self, value: t.Optional[bytes], expression, connection):
             return None
 
         # Wrap it so __set__ knows this is NOT new user input.
-        return _TrustedCiphertext(value)
+        return _TrustedCiphertext(value, _TrustedCiphertext.Source.DB)
 
     def pre_save(
         self, model_instance: EncryptedModel, add  # type: ignore[override]

codeforlife/models/fields/base_encrypted_test.py

@@ -84,8 +84,8 @@ def _value_to_bytes(value: str):
     def _bytes_to_value(data: bytes):
         return data.decode()
 
-    def __init__(self, associated_data, default=None, **kwargs):
-        super().__init__(associated_data, default, **kwargs)
+    def __init__(self, associated_data, **kwargs):
+        super().__init__(associated_data, **kwargs)
 
         self.value_to_bytes = MagicMock(side_effect=self._value_to_bytes)
         self.bytes_to_value = MagicMock(side_effect=self._bytes_to_value)
@@ -148,14 +148,12 @@ def test_init__no_associated_data(self):
     def test_init(self):
         """BaseEncryptedField is constructed correctly."""
         assert self.field.associated_data == self.field_associated_data
-        assert self.field.db_column == self.field_associated_data
 
     def test_deconstruct(self):
         """BaseEncryptedField is deconstructed correctly."""
         _, _, _, kwargs = self.field.deconstruct()
 
         assert kwargs["associated_data"] == self.field_associated_data
-        assert kwargs["db_column"] == self.field_associated_data
 
     # --------------------------------------------------------------------------
     # Django Model Field Integration Tests
@@ -318,7 +316,9 @@ def test_set__none(self):
 
     def test_set__trusted_ciphertext(self):
         """Setting field to _TrustedCiphertext stores ciphertext directly."""
-        trusted_ciphertext = _TrustedCiphertext(b"encrypted_value")
+        trusted_ciphertext = _TrustedCiphertext(
+            b"encrypted_value", _TrustedCiphertext.Source.DB
+        )
         instance = self._get_model_instance(field=trusted_ciphertext)
         assert instance.get_stored_value(self.field) is trusted_ciphertext
 
@@ -381,7 +381,10 @@ def test_get__descriptor(self):
     def test_get__cached(self):
         """Getting field when cached returns cached value."""
         instance = self._get_model_instance()
-        instance.set_stored_value(self.field, _TrustedCiphertext(b"irrelevant"))
+        instance.set_stored_value(
+            self.field,
+            _TrustedCiphertext(b"irrelevant", _TrustedCiphertext.Source.DB),
+        )
 
         value = "decrypted_value"
         instance.__decrypted_values__[self.field.attname] = value
@@ -414,7 +417,10 @@ def test_get__decrypted_value(self):
 
         # Create instance with stored ciphertext.
         instance = self._get_model_instance()
-        instance.set_stored_value(self.field, _TrustedCiphertext(ciphertext))
+        instance.set_stored_value(
+            self.field,
+            _TrustedCiphertext(ciphertext, _TrustedCiphertext.Source.DB),
+        )
         # Ensure cache is not set initially.
         assert self.field.attname not in instance.__decrypted_values__
 
@@ -478,7 +484,9 @@ def test_pre_save__trusted_ciphertext(self):
         """pre_save with trusted ciphertext does nothing."""
         # Create instance with trusted ciphertext.
         ciphertext = b"encrypted_value"
-        trusted_ciphertext = _TrustedCiphertext(ciphertext)
+        trusted_ciphertext = _TrustedCiphertext(
+            ciphertext, _TrustedCiphertext.Source.DB
+        )
         instance = self._get_model_instance(field=trusted_ciphertext)
 
         # Assert pre_save returns the ciphertext directly.