Add Dataspace FK validation on Dataspace and DejacodeUser models

Signed-off-by: tdruez <[email protected]>

Author: tdruez

dje/models.py

@@ -22,7 +22,6 @@
 from django.contrib.admin.models import CHANGE
 from django.contrib.admin.models import DELETION
 from django.contrib.admin.models import LogEntry
-from django.contrib.auth import get_user_model
 from django.contrib.auth.models import AbstractUser
 from django.contrib.auth.models import BaseUserManager
 from django.contrib.auth.models import Group
@@ -90,6 +89,41 @@ def is_content_type_related(model_class):
     )
 
 
+class DataspaceForeignKeyValidationMixin:
+    """Mixin that enforces all related objects share the same Dataspace as self."""
+
+    def save(self, *args, **kwargs):
+        """Validate all foreign keys share the same Dataspace before saving."""
+        self._validate_fk_dataspace()
+        super().save(*args, **kwargs)
+
+    def _validate_fk_dataspace(self):
+        """Check that all foreign key objects share this instance's Dataspace."""
+        # For these model classes, related objects can still be saved even if
+        # they have a dataspace which is not the current one.
+        allowed_models = [Dataspace, DejacodeUser, ContentType]
+
+        for fk_field in self.local_foreign_fields:
+            if fk_field.related_model in allowed_models:
+                continue
+
+            fk_field_value = getattr(self, fk_field.name, None)
+            if fk_field_value and fk_field_value.dataspace != self.dataspace:
+                raise ValueError(
+                    f'Foreign key field "{fk_field.name}": related object "{fk_field_value}" '
+                    f'has Dataspace "{fk_field_value.dataspace}", expected "{self.dataspace}"'
+                )
+
+    def _get_local_foreign_fields(self):
+        """
+        Return a list of ForeignKey type fields of the model.
+        GenericForeignKey are not included, filtered out with field.concrete
+        """
+        return [field for field in self._meta.get_fields() if field.many_to_one and field.concrete]
+
+    local_foreign_fields = property(_get_local_foreign_fields)
+
+
 class DataspaceManager(models.Manager):
     def get_by_natural_key(self, name):
         return self.get(name=name)
@@ -414,7 +448,7 @@ def tab_permissions_enabled(self):
         return bool(self.get_configuration("tab_permissions"))
 
 
-class DataspaceConfiguration(models.Model):
+class DataspaceConfiguration(DataspaceForeignKeyValidationMixin, models.Model):
     dataspace = models.OneToOneField(
         to="dje.Dataspace",
         on_delete=models.CASCADE,
@@ -756,7 +790,7 @@ def exclude_locked_products(self):
         )
 
 
-class DataspacedModel(models.Model):
+class DataspacedModel(DataspaceForeignKeyValidationMixin, models.Model):
     """Abstract base model for all models that are keyed by Dataspace."""
 
     dataspace = models.ForeignKey(
@@ -848,19 +882,6 @@ def save(self, *args, **kwargs):
         # It needs to be poped before calling the super().save()
         kwargs.pop("copy", None)
 
-        # For these model classes, related objects can still be saved even if
-        # they have a dataspace which is not the current one.
-        allowed_models = [Dataspace, get_user_model(), ContentType]
-
-        for field in self.local_foreign_fields:
-            if field.related_model not in allowed_models:
-                attr_value = getattr(self, field.name)
-                if attr_value and attr_value.dataspace != self.dataspace:
-                    raise ValueError(
-                        f'The Dataspace of the related object: "{attr_value}" '
-                        f'is not "{self.dataspace}"'
-                    )
-
         self.clean_extra_spaces_in_identifier_fields()
         super().save(*args, **kwargs)
 
@@ -1089,15 +1110,6 @@ def urn_link(self):
         if urn:
             return format_html('<a href="{}">{}</a>', reverse("urn_resolve", args=[urn]), urn)
 
-    def _get_local_foreign_fields(self):
-        """
-        Return a list of ForeignKey type fields of the model.
-        GenericForeignKey are not included, filtered out with field.concrete
-        """
-        return [field for field in self._meta.get_fields() if field.many_to_one and field.concrete]
-
-    local_foreign_fields = property(_get_local_foreign_fields)
-
     @classmethod
     def get_identifier_fields(cls, *args, **kwargs):
         """
@@ -1678,7 +1690,7 @@ def get_vulnerability_notifications_users(self, dataspace):
         )
 
 
-class DejacodeUser(AbstractUser):
+class DejacodeUser(DataspaceForeignKeyValidationMixin, AbstractUser):
     uuid = models.UUIDField(
         _("UUID"),
         default=uuid.uuid4,

dje/tests/test_models.py

@@ -20,6 +20,7 @@
 from dje.models import get_unsecured_manager
 from dje.models import is_dataspace_related
 from dje.models import is_secured
+from dje.tests import create
 from organization.models import Owner
 from organization.models import Subowner
 
@@ -222,6 +223,28 @@ def test_dataspace_has_configuration(self):
         DataspaceConfiguration.objects.create(dataspace=self.dataspace)
         self.assertTrue(self.dataspace.has_configuration)
 
+    def test_dataspace_configuration_model_foreign_key_validation(self):
+        layout_dataspace = create("CardLayout", self.dataspace)
+        layout_alternate = create("CardLayout", self.alternate_dataspace)
+
+        expected_message = 'has Dataspace "alternate", expected "nexB"'
+        with self.assertRaisesMessage(ValueError, expected_message):
+            DataspaceConfiguration.objects.create(
+                dataspace=self.dataspace,
+                homepage_layout=layout_alternate,
+            )
+
+        with self.assertRaisesMessage(ValueError, expected_message):
+            self.dataspace.set_configuration("homepage_layout", layout_alternate)
+
+        self.dataspace.set_configuration("homepage_layout", layout_dataspace)
+
+        configuration = self.dataspace.get_configuration()
+
+        with self.assertRaisesMessage(ValueError, expected_message):
+            configuration.homepage_layout = layout_alternate
+            configuration.save()
+
     def test_dataspace_tab_permissions_enabled(self):
         self.assertFalse(self.dataspace.tab_permissions_enabled)
 

dje/tests/test_user.py

@@ -627,6 +627,7 @@ def test_user_password_reset_flow(self):
 class DejaCodeUserModelTestCase(TestCase):
     def setUp(self):
         self.dataspace = Dataspace.objects.create(name="nexB")
+        self.alternate_dataspace = Dataspace.objects.create(name="alternate")
 
     def test_user_model_queryset_manager(self):
         active = create_user("active", self.dataspace)
@@ -725,3 +726,18 @@ def test_user_model_serialize_user_data(self):
             "dataspace": "nexB",
         }
         self.assertEqual(expected, user.serialize_user_data())
+
+    def test_user_model_foreign_key_validation(self):
+        layout_dataspace = create("CardLayout", self.dataspace)
+        layout_alternate = create("CardLayout", self.alternate_dataspace)
+
+        expected_message = 'has Dataspace "alternate", expected "nexB"'
+        with self.assertRaisesMessage(ValueError, expected_message):
+            create_user("active", self.dataspace, homepage_layout=layout_alternate)
+
+        user = create_user("active", self.dataspace, homepage_layout=layout_dataspace)
+        self.assertTrue(user.id)
+
+        with self.assertRaisesMessage(ValueError, expected_message):
+            user.homepage_layout = layout_alternate
+            user.save()