Add Dataspace FK validation on Dataspace and DejacodeUser models (#431)

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

Author: tdruez

CHANGELOG.rst

@@ -17,6 +17,15 @@ Release notes
   Prior to this change only "superusers" could see and use this feature.
   https://github.com/aboutcode-org/dejacode/issues/385
 
+- Add Dataspace FK validation on Dataspace and DejacodeUser models.
+  Assigning an object from another Dataspace will raise an error at the ``save()``
+  level.
+  Do not include the ``homepage_layout`` field on Dataspace "addition" form since the
+  Dataspace does not exist yet.
+  Display the ``homepage_layout`` field as read-only on the Dataspace and User change
+  forms when the currently logged user is not looking at his own Dataspace.
+  https://github.com/aboutcode-org/dejacode/issues/428
+
 ### Version 5.4.2
 
 - Migrate the LDAP testing from using mockldap to slapdtest.

dje/admin.py

@@ -1081,11 +1081,7 @@ class DataspaceConfigurationInline(DataspacedFKMixin, admin.StackedInline):
     form = DataspaceConfigurationForm
     verbose_name_plural = _("Configuration")
     verbose_name = _("Dataspace configuration")
-    fieldsets = [
-        (
-            "",
-            {"fields": ("homepage_layout",)},
-        ),
+    add_fieldsets = [
         (
             "AboutCode Integrations",
             {
@@ -1142,8 +1138,25 @@ class DataspaceConfigurationInline(DataspacedFKMixin, admin.StackedInline):
             },
         ),
     ]
+    # Do not include the Dataspace related FKs on addition as the Dataspace does not exist yet
+    fieldsets = [("", {"fields": ("homepage_layout",)})] + add_fieldsets
     can_delete = False
 
+    def get_fieldsets(self, request, obj=None):
+        if not obj:
+            return self.add_fieldsets
+        return super().get_fieldsets(request, obj)
+
+    def get_readonly_fields(self, request, obj=None):
+        """Only a user from the current Dataspace can edit Dataspace related FKs."""
+        readonly_fields = super().get_readonly_fields(request, obj)
+
+        dataspace = obj
+        if dataspace and dataspace != request.user.dataspace:
+            readonly_fields += ("homepage_layout",)
+
+        return readonly_fields
+
 
 @admin.register(Dataspace, site=dejacode_site)
 class DataspaceAdmin(
@@ -1433,7 +1446,6 @@ class DejacodeUserAdmin(
     add_fieldsets = (
         (None, {"fields": ("username", "dataspace")}),
         (_("Personal information"), {"fields": ("email", "first_name", "last_name", "company")}),
-        (_("Profile"), {"fields": ("homepage_layout",)}),
         (
             _("Notifications"),
             {
@@ -1448,6 +1460,7 @@ class DejacodeUserAdmin(
         (_("Permissions"), {"fields": ("is_staff", "is_superuser", "groups")}),
     )
     fieldsets = add_fieldsets[:-1] + (
+        (_("Profile"), {"fields": ("homepage_layout",)}),
         (_("Permissions"), {"fields": ("is_active", "is_staff", "is_superuser", "groups")}),
         (_("Important dates"), {"fields": ("last_login", "last_api_access", "date_joined")}),
     )
@@ -1516,6 +1529,15 @@ def formfield_for_foreignkey(self, db_field, request=None, **kwargs):
 
         return super().formfield_for_foreignkey(db_field, request, **kwargs)
 
+    def get_readonly_fields(self, request, obj=None):
+        """Only a user from the current Dataspace can edit Dataspace related FKs."""
+        readonly_fields = super().get_readonly_fields(request, obj)
+
+        if obj and obj.dataspace != request.user.dataspace:
+            readonly_fields += ("homepage_layout",)
+
+        return readonly_fields
+
     def user_change_password(self, request, id, form_url=""):
         """
         Remove the possibility to force a new password on a User.

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_admin.py

@@ -188,6 +188,35 @@ def test_dataspace_admin_changeform_update_packages_from_scan_field_validation(s
         response = self.client.post(url, data, follow=True)
         self.assertContains(response, "was changed successfully.")
 
+    def test_dataspace_admin_changeform_hide_dataspace_fk_on_addition(self):
+        self.client.login(username=self.super_user.username, password="secret")
+
+        expected1 = "homepage_layout"
+        expected2 = "Homepage layout"
+
+        url = reverse("admin:dje_dataspace_add")
+        response = self.client.get(url)
+        self.assertNotContains(response, expected1)
+        self.assertNotContains(response, expected2)
+
+        url = reverse("admin:dje_dataspace_change", args=[self.other_dataspace.pk])
+        response = self.client.get(url)
+        self.assertContains(response, expected1)
+        self.assertContains(response, expected2)
+
+    def test_dataspace_admin_changeform_dataspace_fk_read_only_when_other_dataspace(self):
+        self.client.login(username=self.super_user.username, password="secret")
+
+        expected = "configuration-0-homepage_layout"
+
+        url = reverse("admin:dje_dataspace_change", args=[self.other_dataspace.pk])
+        response = self.client.get(url)
+        self.assertNotContains(response, expected)
+
+        url = reverse("admin:dje_dataspace_change", args=[self.dataspace.pk])
+        response = self.client.get(url)
+        self.assertContains(response, expected)
+
     def test_dataspace_admin_changelist_missing_in_filter_availability(self):
         # MissingInFilter is only available to superusers
         url = reverse("admin:organization_owner_changelist")

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

@@ -413,6 +413,34 @@ def test_user_admin_changeform_submit_row_delete_button_label(self):
         )
         self.assertContains(response, expected, html=True)
 
+    def test_user_admin_hide_dataspace_fk_on_addition(self):
+        self.client.login(username="nexb_user", password="secret")
+        url = reverse("admin:dje_dejacodeuser_add")
+        response = self.client.get(url)
+
+        expected1 = "homepage_layout"
+        expected2 = "Homepage layout"
+        self.assertNotContains(response, expected1)
+        self.assertNotContains(response, expected2)
+
+        url = reverse("admin:dje_dejacodeuser_change", args=[self.other_user.pk])
+        response = self.client.get(url)
+        self.assertContains(response, expected1)
+        self.assertContains(response, expected2)
+
+    def test_user_admin_dataspace_fk_read_only_when_other_dataspace(self):
+        self.client.login(username=self.nexb_user.username, password="secret")
+
+        expected = "id_homepage_layout"
+
+        url = reverse("admin:dje_dejacodeuser_change", args=[self.other_user.pk])
+        response = self.client.get(url)
+        self.assertNotContains(response, expected)
+
+        url = reverse("admin:dje_dejacodeuser_change", args=[self.nexb_user.pk])
+        response = self.client.get(url)
+        self.assertContains(response, expected)
+
     def test_user_admin_form_scope_homepage_layout_choices(self):
         self.client.login(username=self.nexb_user.username, password="secret")
         url = reverse("admin:dje_dejacodeuser_change", args=[self.nexb_user.pk])
@@ -627,6 +655,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 +754,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()