validate crypto command (#390)

* allow to write to tar file.

Add a switch to the scionlab-config script to just write the config
to a tar file.
Fix the --tar switch behavior so that it opens a tar file.

* refactor verify functions

* add crypto verification functions to AS and ISD.

These functions check the validity of the current certificates and keys,
as well as the whole chain of TRCs that is stored for the ISD.

* update scion-pki so we can use --check-time

* simple command to check validity of crypto.

For the moment, it retrieves all ISDs and checks their crypto validity.
The same for all infrastructure ASes.

* update UTs

* from review, changes to scionlab-config

* cleanup

Some functions and a file rename.

* reaname noinstall option to onlydownload.

The argparse option noinstall is now called onlydownload.

* Change verify_ functions' signatures.

Requested from review #390, see comments.

* move cert/key cardinality check to AS.validate_crypto

Author: juagargi

scionlab/hostfiles/scionlab-config

@@ -37,6 +37,7 @@ import shlex
 import shutil
 import subprocess
 import sys
+import pathlib
 import tarfile
 import textwrap
 import time
@@ -109,13 +110,17 @@ def main(argv):
             try:
                 fetch_info = get_fetch_info(args)
                 config = fetch_config(fetch_info)
+                if args.onlydownload:
+                    return write_tarfile(args.onlydownload, config)
+
                 if config is _CONFIG_EMPTY:
                     stop_scion()
                 elif config is _CONFIG_UNCHANGED:
                     logging.info('Configuration unchanged (version %s). Nothing to do.',
                                  fetch_info.version)
                 else:
-                    install_config(args, config)
+                    tgz = tarfile.open(mode='r:gz', fileobj=io.BytesIO(config))
+                    install_config(args, tgz)
                     confirm_deployed(args)
             except TemporaryError as e:
                 if args.daemon:
@@ -181,10 +186,18 @@ def parse_command_line_args(argv):
                                     'locally modified configuration files. '
                                     'This is enabled by default if --force is not set and '
                                     'the stdin is not a TTY.')
-    group_config.add_argument('--tar',
-                              type=argparse.FileType('r'),
-                              help='Install configuration from a tar-file already obtained from '
-                                   'the SCIONLab coordination service')
+    withfile_or_onlydownload = group_config.add_mutually_exclusive_group()
+    withfile_or_onlydownload.add_argument(
+        '--tar',
+        type=pathlib.Path,
+        help='Install configuration from a tar-file already obtained from '
+        'the SCIONLab coordination service')
+    withfile_or_onlydownload.add_argument(
+        '--onlydownload',
+        metavar='FILE',
+        type=pathlib.Path,
+        help='Only obtain the tar-file from the coordinator service, '
+        'write it to the specified path and do not install.')
     group_fetch = parser.add_argument_group('SCIONLab API options')
     group_fetch.add_argument('--host-id', help='Host identifier')
     group_fetch.add_argument('--host-secret', help='Authentication for host')
@@ -243,7 +256,7 @@ def _load_fetch_info(file, args):
     if args.url:
         fetch_info = fetch_info._replace(url=args.url)
 
-    if args.force:
+    if args.force or args.onlydownload:
         fetch_info = fetch_info._replace(version=None)
 
     return fetch_info
@@ -301,7 +314,7 @@ def fetch_config(fetch_info):
         code = conn.getcode()
         if code == 200:
             response_data = conn.read()
-            return tarfile.open(mode='r:gz', fileobj=io.BytesIO(response_data))
+            return response_data
         elif code == 204:
             return _CONFIG_EMPTY
         else:
@@ -321,6 +334,18 @@ def fetch_config(fetch_info):
         raise TemporaryError(error_msg % (fetch_info.url, e))
 
 
+def write_tarfile(filepath, tarbytes):
+    if tarbytes is _CONFIG_EMPTY:
+        logging.warn("will create an empty tar file")
+        buff = io.BytesIO()
+        tt = tarfile.open(mode='w:gz', fileobj=buff)
+        tt.close()
+        tarbytes = buff.getvalue()
+
+    with open(filepath, 'wb') as f:
+        f.write(tarbytes)
+
+
 def confirm_deployed(args):
     """
     Inform the SCIONLab coordinator of the currently installed version of the configuration. This

scionlab/management/commands/check_crypto.py

@@ -0,0 +1,50 @@
+# Copyright 2021 ETH Zurich
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from django.core.management.base import BaseCommand
+
+from scionlab.models.core import ISD
+from scionlab.scion.pkicommand import ScionPkiError
+
+
+class Command(BaseCommand):
+    help = 'Checks all certificates of all infrastructure ASes for their validity'
+
+    def handle(self, *args, **kwargs):
+        failed_isds = []
+        failed_ases = []
+        isds = ISD.objects.all()
+        for isd in isds:
+            try:
+                num = isd.validate_crypto()
+                print(f'ISD {isd.isd_id}: {isd.label} ; {num} TRCs.')
+            except ScionPkiError as ex:
+                failed_isds.append(isd)
+                print(f'---------------------------\n'
+                      f'ERROR: failed to verify ISD {isd.isd_id}: {ex}'
+                      f'\n-----------------------------')
+                continue
+            ases = isd.ases.filter(owner=None)
+            for as_ in ases:
+                try:
+                    num = as_.validate_crypto()
+                    print(f'AS {as_.as_id}, core: {as_.is_core} ; {num} certs')
+                except ScionPkiError as ex:
+                    failed_ases.append(as_)
+                    print(f'failed AS: {as_.as_id}: {ex}')
+
+        print(f'summary; {len(failed_isds)} failed ISDs: '
+              f'{[isd.isd_id for isd in failed_isds]}')
+        print(f'summary; {len(failed_ases)} failed ASes: '
+              f'{[as_.as_id for as_ in failed_ases]}')

scionlab/models/core.py

@@ -33,6 +33,10 @@
 from scionlab.models.user import User
 from scionlab.models.pki import Key, Certificate
 from scionlab.scion import as_ids
+from scionlab.scion.certs import verify_certificate_valid, verify_cp_as_chain
+from scionlab.scion.keys import verify_key
+from scionlab.scion.trcs import decode_trc, verify_trcs
+from scionlab.scion.pkicommand import ScionPkiError
 from scionlab.util.django import value_set
 from scionlab.defines import (
     MAX_INTERFACE_ID,
@@ -108,6 +112,16 @@ def update_trc_and_certificates(self):
 
         return self.trcs.create()
 
+    def validate_crypto(self):
+        """
+        checks that the crypto material for this object is correct,
+        namely that the trc chain is valid.
+        Returns the number of valid trcs in this ISD.
+        """
+        ts = [decode_trc(t.trc) for t in self.trcs.order_by('serial_version')]
+        verify_trcs(ts[0], *ts)
+        return len(ts)
+
 
 class ASManager(models.Manager):
     def create(self, isd, as_id, is_core=False, label=None, mtu=None, owner=None,
@@ -287,6 +301,33 @@ def keys_latest(self):
                                              version__gte=latest_version)
         return keys
 
+    def validate_crypto(self):
+        """
+        checks that the crypto material for this object is correct, by obtaining the
+        latests certificates and keys and checking them against the latest TRC.
+        Returns the number of certificates (with keys) that passed the check.
+        """
+        trc = decode_trc(self.isd.trcs.latest().trc)
+        certs = self.certificates_latest()
+        cert_map = {}
+        for cert in certs:
+            cert_map[cert.usage()] = cert.certificate
+            verify_certificate_valid(cert.certificate, cert.usage())
+            if cert.usage() == Key.CP_AS:
+                verify_cp_as_chain(cert.format_certfile(), trc)  # chain
+        keys = self.keys_latest()
+        for key in keys:
+            if key.usage not in cert_map:
+                raise ScionPkiError(f'no corresponding cert for key {key.filename()}')
+            verify_key(key.key, cert_map[key.usage])
+        if len(keys) != len(certs):
+            raise ScionPkiError(f'different number of keys ({len(keys)}) than certs ({len(certs)})')
+        if (self.is_core and len(keys) != 5) or (not self.is_core and len(keys) != 1):
+            raise ScionPkiError(
+                f'AS {self.as_id}, core:{self.is_core}: '
+                f'invalid number of keys/certs {len(keys)}')
+        return len(keys)
+
     def generate_keys(self, not_before=None):
         Key.objects.create_all_keys(self, not_before=not_before)
 

scionlab/scion/certs.py

@@ -26,8 +26,11 @@
 from cryptography.hazmat.primitives import serialization, hashes
 from cryptography.hazmat.primitives.asymmetric import ec
 from datetime import datetime
+from tempfile import NamedTemporaryFile
 from typing import List, Tuple, Optional, NamedTuple
 
+from scionlab.scion.pkicommand import run_scion_pki
+
 
 OID_ISD_AS = ObjectIdentifier('1.3.6.1.4.1.55324.1.2.1')
 OID_SENSITIVE_KEY = ObjectIdentifier('1.3.6.1.4.1.55324.1.3.1')
@@ -59,6 +62,35 @@ def decode_certificate(pem: str) -> x509.Certificate:
     return x509.load_pem_x509_certificate(pem.encode("ascii"))
 
 
+def verify_certificate_valid(cert: str, cert_usage: str):
+    """
+    Verifies that the certificate's fields are valid for that type.
+    The certificate is passed as a PEM string.
+    This function does not verify the trust chain
+    (see also verify_cp_as_chain).
+    """
+    with NamedTemporaryFile('w', suffix=".crt") as f:
+        f.write(cert)
+        f.flush()
+        _run_scion_pki_certificate('validate', '--type', cert_usage, '--check-time', f.name)
+
+
+def verify_cp_as_chain(cert: str, trc: bytes):
+    """
+    Verify that the certificate is valid, using the last TRC as anchor.
+    The certificate is passed as a PEM string.
+    The TRC is passed as bytes, basee 64 format.
+    Raises ScionPkiError if the certificate is not valid.
+    """
+    with NamedTemporaryFile(mode='wb', suffix=".trc") as trc_file,\
+         NamedTemporaryFile(mode='wt', suffix=".pem") as cert_file:
+        files = [trc_file, cert_file]
+        for f, value in zip(files, [trc, cert]):
+            f.write(value)
+            f.flush()
+        _run_scion_pki_certificate('verify', '--trc', trc_file.name, cert_file.name)
+
+
 def generate_voting_sensitive_certificate(subject_id: str,
                                           subject_key: ec.EllipticCurvePrivateKey,
                                           not_before: datetime,
@@ -250,3 +282,7 @@ def _build_extensions_as(subject_key: ec.EllipticCurvePrivateKey,
                                              x509.ExtendedKeyUsageOID.CLIENT_AUTH,
                                              x509.ExtendedKeyUsageOID.TIME_STAMPING]),
                       critical=False)]
+
+
+def _run_scion_pki_certificate(*args, cwd=None, check=True):
+    return run_scion_pki('certificate', *args, cwd=cwd, check=check)

scionlab/scion/keys.py

@@ -18,10 +18,15 @@
 ========================================================================
 """
 
+import contextlib
+
 from cryptography.hazmat.primitives.asymmetric import ec
 from cryptography.hazmat.primitives import serialization
+from tempfile import NamedTemporaryFile
 from typing import cast
 
+from scionlab.scion.pkicommand import run_scion_pki
+
 
 def generate_key() -> ec.EllipticCurvePrivateKeyWithSerialization:
     """ Generate an elliptic curve private key """
@@ -43,3 +48,24 @@ def encode_key(key: ec.EllipticCurvePrivateKeyWithSerialization) -> str:
 def decode_key(pem: str) -> ec.EllipticCurvePrivateKey:
     """ Returns an EllipticCurve key from its PEM encoding """
     return serialization.load_pem_private_key(pem.encode("ascii"), password=None)
+
+
+def verify_key(key: str, cert: str):
+    """
+    Verify that the certificate is valid, using the last TRC as anchor.
+    The key is passed as a PEM string.
+    The certificate is passed as a PEM string.
+    Raises ScionPkiError if the certificate is not valid.
+    """
+    with contextlib.ExitStack() as stack:
+        key_file = stack.enter_context(NamedTemporaryFile(mode='wt', suffix=".key"))
+        cert_file = stack.enter_context(NamedTemporaryFile(mode='wt', suffix=".crt"))
+        files = [key_file, cert_file]
+        for f, value in zip(files, [key, cert]):
+            f.write(value)
+            f.flush()
+        _run_scion_pki_key('match', 'certificate', key_file.name, cert_file.name)
+
+
+def _run_scion_pki_key(*args, cwd=None, check=True):
+    return run_scion_pki('key', *args, cwd=cwd, check=check)

scionlab/scion/pkicommand.py

@@ -0,0 +1,52 @@
+# Copyright 2021 ETH Zurich
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+:mod:`scionlab.scion.util` --- scion-pki execution
+==================================================
+"""
+
+import subprocess
+
+from django.conf import settings
+
+
+def run_scion_pki(*args, cwd=None, check=True):
+    try:
+        return subprocess.run([settings.SCION_PKI_COMMAND, *args],
+                              stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                              encoding='utf-8',
+                              check=check,
+                              cwd=cwd)
+    except subprocess.CalledProcessError as e:
+        raise ScionPkiError(e) from None
+
+
+class ScionPkiError(subprocess.CalledProcessError):
+    """
+    Wrapper for CalledProcessError (raised by subprocess.run on returncode != 0 if check=True),
+    that includes the process output (stdout) in the __str__.
+    """
+    def __init__(self, e):
+        if isinstance(e, subprocess.CalledProcessError):
+            super().__init__(e.returncode, e.cmd, e.output, e.stderr)
+        else:
+            super().__init__(returncode=0, cmd='<<internal>>', output=str(e))
+
+    def __str__(self):
+        s = super().__str__()
+        if self.output:
+            s += "\nOutput:\n"
+            s += self.output
+        return s