Allow modifying is_core via admin interface. (#457)

* Allow modifying is_core via admin interface.

If modified, it triggers a new TRC, new keys and certs for all the ISD,
and REMOVES all links that the AS had.

* linting.

* Language.

* Test core and non-core modification.

Author: juagargi

scionlab/admin.py

@@ -16,7 +16,7 @@
 from django import urls
 from django.utils.html import format_html
 from django import forms
-from django.contrib import admin
+from django.contrib import admin, messages
 from django.core.exceptions import ValidationError
 from django.db.models import Count
 from django.shortcuts import get_object_or_404
@@ -485,10 +485,8 @@ def get_readonly_fields(self, request, obj):
         # FIXME(matzf) conceptually, an AS can change the ISD. Not allowed for now
         # as I anticipate this may unnecessarily complicate the TRC/certificate
         # update logic. Should be revisited.
-        # TODO(matzf): Changing is_core should also be possible, not yet implemented
-        #              Requires removing core links etc, bump signed certificates
         if obj:
-            return ('isd', 'is_core', 'as_id',)
+            return ('isd', 'as_id',)
         return ()
 
     def get_inline_instances(self, request, obj):
@@ -533,6 +531,16 @@ def update_core_keys(self, request, queryset):
         """
         AS.update_core_as_keys(queryset)
 
+    def save_model(self, request, obj, form, change):
+        """
+        Custom save model logic, to be able to show a warning message if saving the AS returned one.
+        """
+        # Instead of calling super().save_model(...) we save the object directly and retrieve
+        # its informational object about its changes.
+        msg = obj.save_with_message()
+        if msg is not None:
+            messages.add_message(request, messages.WARNING, msg)
+
 
 class VPNCreationForm(_CreateUpdateModelForm):
     """

scionlab/models/core.py

@@ -333,6 +333,13 @@ def generate_certs(self):
         Certificate.objects.create_all_certs(self)
 
     def update_keys_certs(self, not_before=None):
+        """
+        Updates the keys and certs, and saves this AS object.
+        """
+        self._update_keys_certs(not_before)
+        self.save()
+
+    def _update_keys_certs(self, not_before=None):
         """
         Generate new keys and certificates (core and non core).
         The keys and certs have to be updated simultaneously to avoid getting a nil validity range
@@ -343,7 +350,49 @@ def update_keys_certs(self, not_before=None):
         self.generate_keys(not_before)
         self.generate_certs()
         self.hosts.bump_config()
-        self.save()
+
+    def save(self, **kwargs):
+        # Do not return the message from `save_with_message` when calling save().
+        self.save_with_message(**kwargs)
+
+    def save_with_message(self, **kwargs):
+        """
+        Returns a message to be displayed when interactively modifying this AS.
+        It will be used from the ASAdmin class.
+        """
+        msg = None
+        core_change = False
+        # Check if is_core has changed: load previous object and compare with this one.
+        if self.pk is not None:  # Check if the object already exists in the database
+            # Fetch the current value from the database
+            # old_as = AS.objects.filter(pk=self.pk).values_list('is_core', flat=True).first()
+            old_as = AS.objects.filter(pk=self.pk).first()
+            if old_as.is_core != self.is_core:
+                # This AS' core attribute has changed.
+                # We must delete all existing links.
+                interfaces = self.interfaces.all()
+                links = Link.objects.filter(models.Q(
+                    interfaceA__in=interfaces) | models.Q(interfaceB__in=interfaces))
+                link_count = links.count()
+                links.delete()
+                core_change = True
+                # Finally, return a message to the caller (possibly from ASAdmin) to let them
+                # know about the modifications.
+                msg = f"The CORE property has changed: There was {link_count} link(s) deleted. " + \
+                    "New certificates for this AS, and new TRC for this ISD have been created. " + \
+                    "Please review changes."
+
+        if core_change:
+            # Generate new certificates only for this AS. If this was a core AS and CA for other
+            # ASes, we would have to manually update certificates and keys of those other ASes.
+            self._update_keys_certs()
+        # Save the model.
+        super().save(**kwargs)
+        if core_change:
+            # Because the core AS set has changed, we need a new TRC.
+            self.isd.trcs.create()
+
+        return msg
 
     @staticmethod
     def update_cp_as_keys(queryset):
@@ -364,7 +413,7 @@ def update_cp_as_keys(queryset):
     def update_core_as_keys(queryset):
         """
         All the ASes in the queryset must update their keys and certificates.
-        Since this some of these ASes could act as CA for other, non-core ASes,
+        Since this some of these ASes could now act as CA for other (non-core) ASes,
         we must re-issue all certificates in that ISD.
         So in this function, we just update keys, certificates and TRCs for all
         core ASes in all ISDs that are touched by the given queryset.

scionlab/tests/test_models.py

@@ -14,6 +14,7 @@
 import ipaddress
 
 from unittest.mock import patch
+from django.db import models
 from django.test import TestCase
 from scionlab.defines import (
     DISPATCHER_PORT,
@@ -119,6 +120,82 @@ def test_update_keys(self):
         )
         self.assertEqual(new_certificate.version, prev_certificate.version + 1)
 
+    def test_is_core_not_modified(self):
+        # if the is_core property is not modified, do not expect new keys or certs.
+        as_ = AS.objects.get(as_id='ffaa:0:1304')
+        prev_certs = list(as_.certificates())
+        prev_keys = list(as_.keys.all())
+        as_.mtu = 800  # modify a property but not is_core
+        as_.save()
+
+        got_certs = list(as_.certificates().all())
+        got_keys = list(as_.keys.all())
+        self.assertEqual(got_certs, prev_certs)
+        self.assertEqual(got_keys, prev_keys)
+
+    def test_is_core_modification(self):
+        # Subfunction to perform the repetitive test.
+        def _change_is_core_and_test(as_, core_state: bool):
+            # Store state of ffaa:0:1304.
+            prev_certs = list(as_.certificates())
+            prev_keys = list(as_.keys.all())
+            prev_trcs = list(as_.isd.trcs.all())
+            prev_ifaces = as_.interfaces.all()
+            prev_links = Link.objects.filter(
+                models.Q(interfaceA__in=prev_ifaces) |
+                models.Q(interfaceB__in=prev_ifaces))
+            self.assertGreater(prev_links.count(), 0)
+            prev_links = list(prev_links)
+            prev_ifaces = list(prev_ifaces)
+
+            # now promote AS 19-ffaa:0:1304 to core AS. We expect the following to happen:
+            # - existing links starting or ending at that AS are removed.
+            # - new keys and certificates are issued (as core) for that AS.
+            # - all certificates in that ISD are reissued.
+            # - the ISD issues a new TRC
+            as_.mtu = 800
+            as_.is_core = core_state
+            as_.save()
+
+            # Verify the previous links do not exist anymore.
+            self.assertEqual(Link.objects.filter(pk__in=[
+                link.pk for link in prev_links]).count(), 0)
+
+            # Same with interfaces.
+            self.assertEqual(Interface.objects.filter(pk__in=[
+                iface.pk for iface in prev_ifaces]).count(), 0)
+
+            # This AS doesn't have interfaces or links.
+            got_ifaces = as_.interfaces.all()
+            self.assertEqual(got_ifaces.count(), 0)
+            got_links = Link.objects.filter(
+                models.Q(interfaceA__in=got_ifaces) |
+                models.Q(interfaceB__in=got_ifaces))
+            self.assertEqual(got_links.count(), 0)
+
+            # We have now more certificates and keys.
+            got_certs = list(as_.certificates().all())
+            self.assertTrue(set(prev_certs).issubset(got_certs))
+            got_keys = list(as_.keys.all())
+            self.assertTrue(set(prev_keys).issubset(got_keys))
+
+            # We have a new TRC.
+            got_trcs = list(as_.isd.trcs.all())
+            self.assertTrue(set(prev_trcs).issubset(got_trcs))
+
+        # Get two ASes from the fixture, one core and one non-core.
+        as1 = AS.objects.get(as_id='ffaa:0:1301')
+        as2 = AS.objects.get(as_id='ffaa:0:1304')
+        # test sanity check: the fixture should have:
+        # ffaa:0:1301 core.
+        # ffaa:0:1304 non-core.
+        self.assertTrue(as1.is_core)
+        self.assertFalse(as2.is_core)
+
+        # Test.
+        _change_is_core_and_test(as1, False)
+        _change_is_core_and_test(as2, True)
+
 
 class LinkModificationTests(TestCase):
     fixtures = []