Merge pull request #5388 from c-po/remove-acme-autocert

pki: T9135: derive ACME certificate chains from disk

3 files changed, 101 insertions, 3 deletions

diff --git a/python/vyos/config.py b/python/vyos/config.py
index 7c9771ae3..113136901 100644
--- a/python/vyos/config.py
+++ b/python/vyos/config.py
@@ -359,10 +359,39 @@ class Config(object):
                                             get_first_key=True)
             if pki_dict:
                 if 'certificate' in pki_dict:
+                    from vyos.defaults import directories
+                    from vyos.pki import acme_chain_ca_entry
+                    from vyos.pki import acme_chain_redundant
+                    vyos_certbot_dir = directories['certbot']
+                    # Snapshot of explicitly/manually configured CAs, taken
+                    # before any synthetic entries are added below
+                    real_ca_certs = dict(pki_dict.get('ca', {}))
+
                     for certificate in pki_dict['certificate']:
                         pki_dict['certificate'][certificate] = config_dict_mangle_acme(
                             certificate, pki_dict['certificate'][certificate])
 
+                        # If this is an ACME certificate, also make its
+                        # intermediate CA chain (read live from certbot's
+                        # own chain.pem, never persisted to the CLI)
+                        # available under pki.ca, the same way consumers
+                        # already look up any manually-configured CA, so
+                        # find_chain() can build the full chain for them -
+                        # unless an explicit, manually-configured CA
+                        # already completes the chain, making this
+                        # redundant.
+                        cert_conf = pki_dict['certificate'][certificate]
+                        leaf_cert = cert_conf.get('certificate')
+                        if leaf_cert and 'acme' in cert_conf and not acme_chain_redundant(
+                                leaf_cert, real_ca_certs):
+                            ca_dict = pki_dict.setdefault('ca', {})
+                            for autochain_name, chain_entry in acme_chain_ca_entry(
+                                    vyos_certbot_dir, certificate).items():
+                                # A real, manually-configured CLI CA object
+                                # with this name wins over the synthetic one
+                                if autochain_name not in ca_dict:
+                                    ca_dict[autochain_name] = chain_entry
+
                 conf_dict['pki'] = pki_dict
 
         interfaces_root = root_dict.get('interfaces', {})
diff --git a/python/vyos/pki.py b/python/vyos/pki.py
index 17fa97223..af0dee626 100644
--- a/python/vyos/pki.py
+++ b/python/vyos/pki.py
@@ -470,6 +470,72 @@ def find_chain(cert, ca_certs):
 
     return chain
 
+# ACME certificates are managed entirely by certbot under
+# {vyos_certbot_dir}/live/<cert_name>/ - the intermediate CA certbot
+# obtained alongside the leaf cert (chain.pem) is derived data, not
+# configuration, so it is never written to the CLI. AUTOCHAIN_<cert_name>
+# is only ever used as an in-memory dict key (see
+# Config.get_config_dict()'s with_pki handling and
+# src/op_mode/pki.py's get_config_ca_certificate()) so consumers that
+# build a full certificate chain via find_chain() - and "show pki ca" -
+# see the same data they would if it had been manually configured,
+# without it ever being a real, settable, or deletable CLI object.
+AUTOCHAIN_PREFIX = 'AUTOCHAIN_'
+
+def acme_chain_ca_entry(vyos_certbot_dir: str, cert_name: str) -> dict:
+    """Build synthetic 'pki ca <name>'-shaped dict entries
+    (`{<name>: {'certificate': <base64>}, ...}`) for every certificate in
+    an ACME certificate's chain, read live from certbot's own chain.pem.
+
+    certbot commonly writes more than just the immediate intermediate to
+    chain.pem (e.g. the intermediate's own issuing root too), and every
+    one of them is needed for find_chain() to walk the full chain - a
+    single 'pki ca' object only ever holds one certificate, so each one
+    gets its own synthetic entry here.
+
+    Returns an empty dict if chain.pem is not (yet) present - e.g. the
+    certificate has not been issued yet, or a prior request failed - so
+    callers can skip this certificate's chain gracefully instead of
+    crashing.
+    """
+    import re
+    from vyos.utils.file import read_file
+    tmp = read_file(f'{vyos_certbot_dir}/live/{cert_name}/chain.pem',
+                     defaultonfailure=None)
+    if tmp is None:
+        return {}
+
+    entries = {}
+    for block in re.findall(
+            r'-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----',
+            tmp, re.DOTALL):
+        cert = load_certificate(block, wrap_tags=False)
+        if not cert:
+            continue
+        index = len(entries) + 1
+        name = f'{AUTOCHAIN_PREFIX}{cert_name}' if index == 1 else \
+               f'{AUTOCHAIN_PREFIX}{cert_name}_{index}'
+        chain_base64 = "".join(encode_certificate(cert).strip().split("\n")[1:-1])
+        entries[name] = {'certificate': chain_base64}
+    return entries
+
+def acme_chain_redundant(leaf_cert_base64: str, ca_certs: dict) -> bool:
+    """Return True if one of the already-configured CAs in ca_certs (a
+    'pki ca'-shaped dict, e.g. Config's pki_dict['ca']) directly signs the
+    given leaf certificate - meaning an explicit, manually-configured CA
+    already completes the chain, so synthesizing/showing an
+    AUTOCHAIN_<cert_name> entry (see acme_chain_ca_entry()) is redundant.
+    """
+    leaf_cert = load_certificate(leaf_cert_base64)
+    if not leaf_cert:
+        return False
+    loaded_ca_certs = [
+        load_certificate(ca['certificate'])
+        for ca in ca_certs.values() if 'certificate' in ca
+    ]
+    loaded_ca_certs = [cert for cert in loaded_ca_certs if cert]
+    return find_parent(leaf_cert, loaded_ca_certs) is not None
+
 def sort_ca_chain(ca_names, pki_node):
     def ca_cmp(ca_name1, ca_name2, pki_node):
         cert1 = load_certificate(pki_node[ca_name1]['certificate'])
diff --git a/python/vyos/utils/file.py b/python/vyos/utils/file.py
index 08a3fa82d..5c2eae4d5 100644
--- a/python/vyos/utils/file.py
+++ b/python/vyos/utils/file.py
@@ -31,10 +31,13 @@ def file_is_persistent(path):
     absolute = os.path.abspath(os.path.dirname(path))
     return re.match(location,absolute)
 
-def read_file(fname, defaultonfailure=None, sudo=False):
+_unset = object()
+
+def read_file(fname, defaultonfailure=_unset, sudo=False):
     """
     read the content of a file, stripping any end characters (space, newlines)
-    should defaultonfailure be not None, it is returned on failure to read
+    should defaultonfailure be given (including None), it is returned on
+    failure to read instead of raising
     """
     try:
         # Some files can only be read by root - emulate sudo cat call
@@ -47,7 +50,7 @@ def read_file(fname, defaultonfailure=None, sudo=False):
                 data = f.read()
         return data.strip()
     except Exception as e:
-        if defaultonfailure is not None:
+        if defaultonfailure is not _unset:
             return defaultonfailure
         raise e