Duplication and isolated attribute management.

- Duplcated names now gives 409 instead of 400. Sadly, the default returned from Django on duplicated entries in the database is 400, and we want 409.
- The NetworkPolicyAttribute 'isolated' is now persistent,  created during migration, and may not be renamed, deleted, or otherwise removed. One can however change its description.

docs/network_policies.md

@@ -40,21 +40,26 @@ This document describes the **Network Policies API**, which manages **Network Po
 
 A **Network Policy** is a named entity grouping a set of attributes (via **Policy Attributes**) and referencing any number of **Communities**. Policies are used to configure or categorize network-related aspects of your infrastructure.
 
-- **Name**: Human-readable identifier of the policy.
+- **Name**: Human-readable case insensitive identifier of the policy.
 - **Attributes**: A list of attributes that apply to this policy (see below).
 
 ### Policy Attributes
 
 A **Policy Attribute** is a simple boolean flag or marker. Policy Attributes do *not* by themselves define system behavior. Instead, each attribute is a type of marker that can be set for a given policy.
 
+- **Name**: Human-readable case insensitive identifier of the attribute.
+- **Description**: A brief description of the attribute.
+
+Note: The attribute 'isolated' is a special attribute that is used to mark a network as isolated, this attribute is hardcoded and cannot be deleted, updated or created (but you may still change its description).
+
 **Important**:  
 
 - Attributes must first be defined globally (e.g., "isolated," "public," "private") using the **Policy Attribute** endpoints.  
 - Then, they are associated with a **Network Policy** and given a value (true/false).  
 
 ### Community
 
-A **Community** is a named collection of **Hosts** within a **single** Network Policy. Each community belongs to exactly one policy, enabling further subdivision of hosts adhering to that policy. Communities allow you to group hosts based on function, location, access, or other logical groupings.
+A **Community** is a named (case insensitive) collection of **Hosts** within a **single** Network Policy. Each community belongs to exactly one policy, enabling further subdivision of hosts adhering to that policy. Communities allow you to group hosts based on function, location, access, or other logical groupings.
 
 ### Host Membership
 
@@ -88,7 +93,7 @@ A **Community** is a named collection of **Hosts** within a **single** Network P
 
 ```json
 {
-  "name": "SecurePolicy",
+  "name": "SecurePolicy", // This will be turned into "securepolicy" internally
   "attributes": [
     {
       "attribute": 1,   // ID of an existing NetworkPolicyAttribute

mreg/api/v1/endpoints.py

@@ -10,3 +10,4 @@ class NetworkPolicy(str, Enum):
         COMMUNITY_HOSTS_LIST = "networkpolicy-community-hosts-list"
         COMMUNITY_HOST_DETAIL = "networkpolicy-community-host-detail"
         ATTRIBUTE_LIST = "networkpolicyattribute-list"
+        ATTRIBUTE_DETAIL = "networkpolicyattribute-detail"        

mreg/api/v1/tests/test_network_policy.py

@@ -10,7 +10,6 @@
 POLICY_ENDPOINT = "/api/v1/networkpolicies/"
 ATTRIBUTE_ENDPOINT = "/api/v1/networkpolicyattributes/"
 
-
 class NetworkPolicyTestCase(ParametrizedTestCase, MregAPITestCase):
     def setUp(self):
         super().setUp()
@@ -42,6 +41,104 @@ def _create_community(self, name: str, description: str, policy: NetworkPolicy)
     def _delete_community(self, name: str):
         Community.objects.get(name=name).delete()
 
+    def _get_protected_attribute_isolated(self) -> NetworkPolicyAttribute:
+        return NetworkPolicyAttribute.objects.get(name="isolated")
+
+    def _get_protected_attribute(self, name: str) -> NetworkPolicyAttribute:
+        return NetworkPolicyAttribute.objects.get(name=name)
+
+    def _try_get_protected_attribute(self, name: str) -> NetworkPolicyAttribute | None:
+        try:
+            return self._get_protected_attribute(name)
+        except NetworkPolicyAttribute.DoesNotExist:
+            return None
+
+    def test_isolated_attribute_protected_exists(self):
+        """Test that the isolated attribute is protected and exists."""
+        self.assertIsNotNone(self._get_protected_attribute_isolated())
+
+    def test_create_attribute_protected_409(self):
+        """Test creating a protected network policy attribute."""
+        data = {"name": "isolated", "description": "attribute desc"}
+        self.assert_post_and_409(ATTRIBUTE_ENDPOINT, data=data)
+
+        data["name"] = data["name"].upper()
+        self.assert_post_and_409(ATTRIBUTE_ENDPOINT, data=data)
+
+    def test_delete_attribute_protected_403(self):
+        """Test deleting a protected network policy attribute."""
+        isolated = self._get_protected_attribute_isolated()
+        self.assert_delete_and_403(f"{ATTRIBUTE_ENDPOINT}{isolated.pk}")
+
+    def test_patch_attribute_protected_name_403(self):
+        """Test updating a protected network policy attribute."""
+        isolated = self._get_protected_attribute_isolated()
+        data = {"name": "new_attribute", "description": "new attribute desc"}
+        self.assert_patch_and_403(f"{ATTRIBUTE_ENDPOINT}{isolated.pk}", data=data)
+
+    def test_patch_attribute_protected_description_201(self):
+        """Test updating a protected network policy attribute."""
+        isolated = self._get_protected_attribute_isolated()
+        data = {"description": "new attribute desc"}
+        self.assert_patch_and_200(f"{ATTRIBUTE_ENDPOINT}{isolated.pk}", data=data)
+
+    def test_create_attribute(self):
+        """Test creating a network policy attribute."""
+        data = {"name": "attribute", "description": "attribute desc"}
+        ret = self.assert_post_and_201(ATTRIBUTE_ENDPOINT, data=data)
+        self.assertEqual(ret.json()["name"], "attribute")
+        self.assertEqual(ret.json()["description"], "attribute desc")
+
+        self._delete_attributes(["attribute"])
+
+    def test_create_attribute_duplicate_name_409(self):
+        """Test creating a network policy attribute with a duplicate name."""
+        data = {"name": "attribute", "description": "attribute desc"}
+        self.assert_post_and_201(ATTRIBUTE_ENDPOINT, data=data)
+        self.assert_post_and_409(ATTRIBUTE_ENDPOINT, data=data)
+
+        data["name"] = data["name"].upper()
+        self.assert_post_and_409(ATTRIBUTE_ENDPOINT, data=data)
+
+        self._delete_attributes(["attribute"])
+
+    def test_get_attribute(self):
+        """Test getting a network policy attribute."""
+        data = {"name": "attribute", "description": "attribute desc"}
+        ret = self.assert_post_and_201(ATTRIBUTE_ENDPOINT, data=data)
+        attribute_id = ret.json()["id"]
+
+        ret = self.assert_get(f"{ATTRIBUTE_ENDPOINT}{attribute_id}")
+        self.assertEqual(ret.json()["name"], "attribute")
+        self.assertEqual(ret.json()["description"], "attribute desc")
+
+        self._delete_attributes(["attribute"])
+
+    def test_delete_attribute(self):
+        """Test deleting a network policy attribute."""
+        data = {"name": "attribute", "description": "attribute desc"}
+        ret = self.assert_post_and_201(ATTRIBUTE_ENDPOINT, data=data)
+        attribute_id = ret.json()["id"]
+
+        self.assert_delete_and_204(f"{ATTRIBUTE_ENDPOINT}{attribute_id}")
+
+    def test_patch_attribute(self):
+        """Test updating a network policy attribute."""
+        data = {"name": "attribute", "description": "attribute desc"}
+        ret = self.assert_post_and_201(ATTRIBUTE_ENDPOINT, data=data)
+        attribute_id = ret.json()["id"]
+
+        data = {"name": "new_attribute", "description": "new attribute desc"}
+        ret = self.assert_patch_and_200(f"{ATTRIBUTE_ENDPOINT}{attribute_id}", data=data)
+        self.assertEqual(ret.json()["name"], "new_attribute")
+        self.assertEqual(ret.json()["description"], "new attribute desc")
+
+        self._delete_attributes(["new_attribute"])
+
+    def test_get_attribute_not_exists_404(self):
+        """Test getting a network policy attribute that does not exist."""
+        self.assert_get_and_404(f"{ATTRIBUTE_ENDPOINT}99999999")
+
     @parametrize(
         ("name", "attributes"),
         [
@@ -105,6 +202,18 @@ def test_update_np(self):
         self.assertEqual(ret.json()["name"], "new_name")
         self._delete_network_policy("new_name")
 
+    def test_duplicate_np_name_409(self):
+        """Test creating a network policy with a duplicate name."""
+        name = "policy_duplicate"
+        data = {"name": name, "attributes": []}
+        self.assert_post_and_201(POLICY_ENDPOINT, data=data)
+        # Check exact match
+        self.assert_post_and_409(POLICY_ENDPOINT, data=data)
+        # Check case insensitivity
+        data = {"name": name.upper(), "attributes": []}
+        self.assert_post_and_409(POLICY_ENDPOINT, data=data)
+        self._delete_network_policy(name)
+
     def test_create_community_ok(self):
         """Test creating a community."""
         name = "policy_with_community"
@@ -116,6 +225,12 @@ def test_create_community_ok(self):
         ret = self.assert_post_and_201(f"{POLICY_ENDPOINT}{np.pk}/communities/", data=data)
         community_id = ret.json()["id"]
 
+        location = ret.headers["Location"]
+        self.assertIsNotNone(location, "Location header not set")
+
+        path = f"{POLICY_ENDPOINT}{np.pk}/communities/{community_id}"
+        self.assertTrue(location.endswith(path))
+
         get_res = self.assert_get(f"{POLICY_ENDPOINT}{np.pk}/communities/{community_id}")
         self.assertEqual(get_res.json()["name"], "community")
         self.assertEqual(get_res.json()["description"], "community desc")
@@ -148,6 +263,22 @@ def test_create_community_name_already_exists_in_same_policy_409(self):
         np1.delete()
         np2.delete()
 
+    def test_create_community_duplicate_name_same_policy_409(self):
+        """Test that creating a community with the same name in the same policy fails."""
+        name = "policy_with_community"
+        np = self._create_network_policy(name, [])
+        data = {
+            "name": "community",
+            "description": "community desc",
+        }
+        self.assert_post_and_201(f"{POLICY_ENDPOINT}{np.pk}/communities/", data=data)
+        self.assert_post_and_409(f"{POLICY_ENDPOINT}{np.pk}/communities/", data=data)
+
+        data["name"] = data["name"].upper()    
+        self.assert_post_and_409(f"{POLICY_ENDPOINT}{np.pk}/communities/", data=data)
+
+        self._delete_network_policy(name)
+
     def test_create_community_name_already_exists_in_different_policy_ok(self):
         """Test creating a community with the same name in a different policy."""
         np1 = self._create_network_policy("policy_with_community1", [])

mreg/api/v1/urls.py

@@ -97,5 +97,7 @@
          views_network_policy.NetworkCommunityHostDetail.as_view(), name=URL.NetworkPolicy.COMMUNITY_HOST_DETAIL),
     path("networkpolicyattributes/",
          views_network_policy.NetworkPolicyAttributeList.as_view(), name=URL.NetworkPolicy.ATTRIBUTE_LIST),
+     path("networkpolicyattributes/<int:pk>",
+           views_network_policy.NetworkPolicyAttributeDetail.as_view(), name=URL.NetworkPolicy.ATTRIBUTE_DETAIL),
 
 ]

mreg/api/v1/views_network_policy.py

@@ -33,6 +33,18 @@ class NetworkPolicyList(JSONContentTypeMixin, generics.ListCreateAPIView):
     filterset_class = NetworkPolicyFilterSet
 
     def create(self, request, *args, **kwargs):
+        # Note, we can't use the serializer's is_valid method here because that'll raise a 400 exception
+        # if the data is invalid (even if something exists). We need to catch that and raise a 409 instead.
+        name = request.data.get("name")
+        if not name:
+            raise exceptions.ValidationError("'name' is required.")
+
+        try:
+            NetworkPolicy.objects.get(name=name)
+            raise ValidationError409(detail="NetworkPolicy with this name already exists.")
+        except NetworkPolicy.DoesNotExist:
+            pass
+
         serializer = self.get_serializer(data=request.data)
         serializer.is_valid(raise_exception=True)
         with transaction.atomic():
@@ -68,6 +80,30 @@ class NetworkPolicyAttributeList(JSONContentTypeMixin, generics.ListCreateAPIVie
     ordering_fields = ("id",)
 
 
+    def create(self, request, *args, **kwargs):
+        name = request.data.get("name")
+        if not name:
+            raise exceptions.ValidationError("'name' is required.")
+
+        try:
+            NetworkPolicyAttribute.objects.get(name=name)
+            raise ValidationError409(detail="NetworkPolicyAttribute with this name already exists.")
+        except NetworkPolicyAttribute.DoesNotExist:
+            pass
+
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+
+        with transaction.atomic():
+            network_policy_attribute = serializer.save()
+
+        headers = self.get_success_headers(serializer.data)
+        headers["Location"] = request.build_absolute_uri(
+            reverse(URL.NetworkPolicy.ATTRIBUTE_DETAIL, kwargs={"pk": network_policy_attribute.id})
+        )
+
+        return response.Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)
+
 class NetworkPolicyAttributeDetail(JSONContentTypeMixin, generics.RetrieveUpdateDestroyAPIView):
     queryset = NetworkPolicyAttribute.objects.all()
     serializer_class = NetworkPolicyAttributeSerializer
@@ -83,24 +119,43 @@ def get_queryset(self):
         policy_pk = self.kwargs.get("pk")
         return Community.objects.filter(policy__pk=policy_pk).order_by("id")
 
-    def perform_create(self, serializer):
+    def create(self, request, *args, **kwargs):
         policy_pk = self.kwargs.get("pk")
 
-        # Ensure the NetworkPolicy exists, even though the queryset has already done this
-        # This is in case the NetworkPolicy is deleted between the time the queryset is called and
-        # the time this create is called
+        if not policy_pk:
+            raise exceptions.ValidationError("NetworkPolicy ID is required.")
+
+        # Note, we can't use the serializer's is_valid method here because that'll raise a 400 exception
+        # if the data is invalid (even if something exists). We need to catch that and raise a 409 instead.
+        name = request.data.get("name")
+        if not name:
+            raise exceptions.ValidationError("'name' is required.")
+
         try:
             policy = NetworkPolicy.objects.get(pk=policy_pk)
         except NetworkPolicy.DoesNotExist:  # pragma: no cover
             raise exceptions.NotFound("NetworkPolicy not found.")
 
+        # We do not have to worry about case sensitivity here, as the LowerCaseManager for the model will handle that.
         try:
-            Community.objects.get(name=serializer.validated_data["name"], policy=policy)
+            Community.objects.get(name=name, policy=policy)
             raise ValidationError409(detail="Community with this name already exists.")
         except Community.DoesNotExist:
             pass
+
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
 
-        serializer.save(policy=policy)
+        with transaction.atomic():
+            community = serializer.save(policy=policy)
+        headers = self.get_success_headers(serializer.data)
+
+        # Dynamically generate the Location URL
+        headers["Location"] = request.build_absolute_uri(
+            reverse(URL.NetworkPolicy.COMMUNITY_DETAIL, kwargs={"pk": policy.id, "cpk": community.id})
+        )
+        return response.Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)        
+
 
 
 # Retrieve, update, or delete a specific Community under a specific Network

mreg/migrations/0014_network_policy.py

@@ -4,6 +4,13 @@
 import mreg.fields
 from django.db import migrations, models
 
+def create_protected_attributes(apps, schema_editor):
+    """Ensures the 'isolated' attribute is always created."""
+    NetworkPolicyAttribute = apps.get_model("mreg", "NetworkPolicyAttribute")
+    NetworkPolicyAttribute.objects.get_or_create(
+        name="isolated",
+        defaults={"description": "The network uses client isolation"},
+    )
 
 class Migration(migrations.Migration):
 
@@ -51,6 +58,7 @@ class Migration(migrations.Migration):
                 'abstract': False,
             },
         ),
+        migrations.RunPython(create_protected_attributes),
         migrations.AddField(
             model_name='host',
             name='network_community',

mreg/models/network_policy.py

@@ -1,25 +1,47 @@
 from django.db import models
 from mreg.models.base import BaseModel
 from mreg.fields import LowerCaseCharField
+from mreg.managers import LowerCaseManager
 
+from rest_framework import exceptions
 
 class NetworkPolicyAttribute(BaseModel):
     """
     Represents an attribute that can be applied to a NetworkPolicy.
     """
 
+    objects = LowerCaseManager()
+
+    PROTECTED_ATTRIBUTES = {"isolated"}
+
     name = LowerCaseCharField(max_length=100, unique=True)
     description = models.TextField(blank=True, help_text="Description of the attribute.")
 
+    def save(self, *args, **kwargs):
+        if self.pk:
+            original = NetworkPolicyAttribute.objects.filter(pk=self.pk).first()
+            if original and original.name in self.PROTECTED_ATTRIBUTES and self.name != original.name:
+                raise exceptions.PermissionDenied(detail=f"Cannot rename protected attribute '{original.name}'.")
+
+        super().save(*args, **kwargs)
+
+    def delete(self, *args, **kwargs):
+        if self.name in self.PROTECTED_ATTRIBUTES:
+            raise exceptions.PermissionDenied(detail=f"Cannot delete the attribute '{self.name}', it is protected.")
+        super().delete(*args, **kwargs)
+
     def __str__(self):
         return self.name
 
 
+
 class NetworkPolicy(BaseModel):
     """
     Represents a network policy which consists of a set of NetworkPolicyAttributes.
     """
 
+    objects = LowerCaseManager()
+
     name = LowerCaseCharField(max_length=100, unique=True, help_text="Name of the network policy.")
     description = models.TextField(blank=True, help_text="Description of the network policy.")
     attributes = models.ManyToManyField(
@@ -58,6 +80,8 @@ class Community(BaseModel):
     Represents a community within a NetworkPolicy. Hosts can belong to a community.
     """
 
+    objects = LowerCaseManager()
+
     name = LowerCaseCharField(max_length=100, help_text="Policy-unique name of the community.")
     description = models.CharField(blank=True, max_length=250, help_text="Description of the community.")
     policy = models.ForeignKey(