Fix issue when updating TRC, deleting or adding core AS (#387)

* Fix issue when updating TRC, deleting or adding core AS

Fix issue with overwritting certificate with random version of certificate when updating TRC, after changes to the set of core ASes
Add regression test and unit test checking the new behavior of bundling the latest cert version and not some cert version

* Lint, update comments

* relint

Author: FR4NK-W

scionlab/models/core.py

@@ -264,8 +264,18 @@ def isd_as_path_str(self):
         return self.isd_as_str().replace(":", "_")
 
     def certificates(self):
-        """ returns a queryset of all of this AS' certificates """
-        return Certificate.objects.filter(key__AS=self)
+        """ returns a queryset of all of this AS' certificates (in random order) """
+        return Certificate.objects.filter(key__AS=self).order_by('?')
+
+    def certificates_latest(self):
+        """ returns a queryset of all the latest certificates of this AS """
+        certs = Certificate.objects.none()
+        for key_usage in Key.USAGES:
+            latest_version = self.keys.filter(usage=key_usage).aggregate(models.Max('version'))[
+                                 'version__max'] or 1
+            certs = certs | Certificate.objects.filter(key__AS=self, key__usage=key_usage,
+                                                       key__version__gte=latest_version)
+        return certs
 
     def generate_keys(self, not_before=None):
         Key.objects.create_all_keys(self, not_before=not_before)
@@ -594,14 +604,14 @@ def active(self):
         """
         return list of Interfaces from active links
         """
-        return self.filter(link_as_interfaceA__active=True) |\
+        return self.filter(link_as_interfaceA__active=True) | \
             self.filter(link_as_interfaceB__active=True)
 
     def inactive(self):
         """
         return list of Interfaces from inactive links
         """
-        return self.filter(link_as_interfaceA__active=False) |\
+        return self.filter(link_as_interfaceA__active=False) | \
             self.filter(link_as_interfaceB__active=False)
 
 

scionlab/scion/config.py

@@ -125,7 +125,7 @@ def _write_trcs(self, dir):
             self.archive.write_bytes((dir, CERT_DIR, trc.filename()), trc.format_trcfile())
 
     def _write_certs(self, dir):
-        for cert in self.AS.certificates().all():
+        for cert in self.AS.certificates_latest().all():
             self.archive.write_text((dir, CRYPTO_DIR, cert.subdir(), cert.filename()),
                                     cert.format_certfile())
 

scionlab/scion/trcs.py

@@ -63,6 +63,21 @@ def verify_trcs(*trcs: bytes):
         _run_scion_pki('verify', '--anchor', *[f.name for f in files])
 
 
+def verify_certificate(cert: bytes, trc: bytes):
+    """
+    Verify that the certificate is valid, using the last TRC as anchor.
+    Raises VerifyError if the certificate is not valid.
+    """
+    with contextlib.ExitStack() as stack:
+        trc_file = stack.enter_context(NamedTemporaryFile(suffix=".trc"))
+        cert_file = stack.enter_context(NamedTemporaryFile(suffix=".pem"))
+        files = [trc_file, cert_file]
+        for f, value in zip(files, [trc, cert]):
+            f.write(value)
+            f.flush()
+        _run_scion_pki('verify', '--trc', *[f.name for f in files], command='certificate')
+
+
 def generate_trc(prev_trc: bytes,
                  isd_id: int,
                  base: int,
@@ -256,13 +271,20 @@ def _to_seconds(d: timedelta) -> str:
         return f'{int(d.total_seconds())}s'
 
 
-def _run_scion_pki(*args, cwd=None, check=True):
+def _run_scion_pki(*args, cwd=None, command='trcs', check=True):
     try:
-        return subprocess.run([settings.SCION_PKI_COMMAND, 'trcs', *args],
-                              stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
-                              encoding='utf-8',
-                              check=check,
-                              cwd=cwd)
+        if command == 'trcs':
+            return subprocess.run([settings.SCION_PKI_COMMAND, command, *args],
+                                  stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                                  encoding='utf-8',
+                                  check=check,
+                                  cwd=cwd)
+        elif command == 'certificate':
+            return subprocess.run([settings.SCION_PKI_COMMAND, command, *args],
+                                  stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                                  encoding='utf-8',
+                                  check=check,
+                                  cwd=cwd)
     except subprocess.CalledProcessError as e:
         raise _CalledProcessErrorWithOutput(e) from None
 

scionlab/tests/test_trc.py

@@ -21,7 +21,6 @@
 from scionlab.models.pki import Key, Certificate
 from scionlab.models.trc import TRC, _coreas_certificates
 
-
 _ASID_1 = 'ff00:0:1'
 _ASID_2 = 'ff00:0:2'
 _ASID_3 = 'ff00:0:3'
@@ -250,6 +249,74 @@ def test_create_sensitive_update(self):
         self.assertTrue(trc.votes.exists())
         self.assertEqual(trc.quorum, prev.quorum)
 
+    def test_delete_one_core_as(self):
+        self._create_ases()
+        prev = TRC.objects.create(self.isd1)
+        # remove one core AS
+        AS.objects.filter(is_core=True, isd=self.isd1).first().delete()
+        # deleting a core As triggers a generation of a TRC. Get that TRC:
+        trc = TRC.objects.latest()
+
+        # check the trc chain
+        _check_trc(trc, prev)
+        # check it's a sensitive update
+        self.assertEqual(trc.serial_version, prev.serial_version + 1)
+        self.assertEqual(trc.base_version, prev.base_version)
+        self.assertEqual(trc.predecessor_trc_or_none(), prev)
+        self.assertTrue(trc.votes.exists())
+        self.assertNotEqual(trc.quorum, prev.quorum)
+
+        # Check valid latest CP AS certificates regenerated, core
+        some_core = AS.objects.filter(is_core=True, isd=self.isd1).first()
+        cert_cp_as = some_core.certificates_latest().filter(key__usage=Key.CP_AS).first()
+        loaded_certs = bytes(cert_cp_as.format_certfile(), 'ascii')
+        trcs.verify_certificate(loaded_certs, trcs.decode_trc(trc.trc))
+
+        # Check valid latest CP AS certificates regenerated, non-core
+        any_none_core = AS.objects.filter(is_core=False, isd=self.isd1).first()
+        cert_cp_as = any_none_core.certificates_latest().filter(key__usage=Key.CP_AS).first()
+        loaded_certs = bytes(cert_cp_as.format_certfile(), 'ascii')
+        trcs.verify_certificate(loaded_certs, trcs.decode_trc(trc.trc))
+
+    def test_broken_delete_one_core_as(self):
+        # [regression test] Check that validating an invalid / old certificate fails
+        # against an updated TRC
+        self._create_ases()
+        prev = TRC.objects.create(self.isd1)
+        # remove one core AS
+        AS.objects.filter(is_core=True, isd=self.isd1).first().delete()
+        # deleting a core As triggers a generation of a TRC. Get that TRC:
+        trc = TRC.objects.latest()
+
+        # check the trc chain
+        _check_trc(trc, prev)
+        # check it's a sensitive update
+        self.assertEqual(trc.serial_version, prev.serial_version + 1)
+        self.assertEqual(trc.base_version, prev.base_version)
+        self.assertEqual(trc.predecessor_trc_or_none(), prev)
+        self.assertTrue(trc.votes.exists())
+        self.assertNotEqual(trc.quorum, prev.quorum)
+
+        # Check invalid CP AS certificates when selecting old certificate, core
+        with self.assertRaises(trcs._CalledProcessErrorWithOutput):
+            some_core = AS.objects.filter(is_core=True, isd=self.isd1).first()
+            cert_cp_as = Certificate.objects.filter(key__AS=some_core, key__usage=Key.CP_AS,
+                                                    key__version=1).get()
+            loaded_certs = bytes(cert_cp_as.format_certfile(), 'ascii')
+            trcs.verify_certificate(loaded_certs, trcs.decode_trc(trc.trc))
+
+        # Check invalid CP AS certificates when randomly selecting, non-core
+        with self.assertRaises(AttributeError):
+            any_none_core = AS.objects.filter(is_core=False, isd=self.isd1).first()
+            cert_cp_as = Certificate.objects.filter(key__AS=any_none_core, key__usage=Key.CP_AS,
+                                                    key__version=1).get()
+            certfile = cert_cp_as.format_certfile()
+            # We should never get further, Unreachable code
+            # The first core AS was deleted and the non-core v1 CP AS cert was referring to
+            # that core AS CA cert
+            loaded_certs = bytes(certfile, 'ascii')
+            trcs.verify_certificate(loaded_certs, trcs.decode_trc(trc.trc))
+
     def test_create_less_core_ases(self):
         self._create_ases()
         prev = TRC.objects.create(self.isd1)