Merge pull request #1380 from sarthurdev/ovpn-multi-ca

openvpn: T4485: Accept multiple tls ca-certificate values

2 files changed, 63 insertions, 2 deletions

diff --git a/python/vyos/pki.py b/python/vyos/pki.py
index fd91fc9bf..cd15e3878 100644
--- a/python/vyos/pki.py
+++ b/python/vyos/pki.py
@@ -332,6 +332,54 @@ def verify_certificate(cert, ca_cert):
     except InvalidSignature:
         return False
 
+def verify_crl(crl, ca_cert):
+    # Verify CRL was signed by specified CA
+    if ca_cert.subject != crl.issuer:
+        return False
+
+    ca_public_key = ca_cert.public_key()
+    try:
+        if isinstance(ca_public_key, rsa.RSAPublicKeyWithSerialization):
+            ca_public_key.verify(
+                crl.signature,
+                crl.tbs_certlist_bytes,
+                padding=padding.PKCS1v15(),
+                algorithm=crl.signature_hash_algorithm)
+        elif isinstance(ca_public_key, dsa.DSAPublicKeyWithSerialization):
+            ca_public_key.verify(
+                crl.signature,
+                crl.tbs_certlist_bytes,
+                algorithm=crl.signature_hash_algorithm)
+        elif isinstance(ca_public_key, ec.EllipticCurvePublicKeyWithSerialization):
+            ca_public_key.verify(
+                crl.signature,
+                crl.tbs_certlist_bytes,
+                signature_algorithm=ec.ECDSA(crl.signature_hash_algorithm))
+        else:
+            return False # We cannot verify it
+        return True
+    except InvalidSignature:
+        return False
+
+def verify_ca_chain(sorted_names, pki_node):
+    if len(sorted_names) == 1: # Single cert, no chain
+        return True
+
+    for name in sorted_names:
+        cert = load_certificate(pki_node[name]['certificate'])
+        verified = False
+        for ca_name in sorted_names:
+            if name == ca_name:
+                continue
+            ca_cert = load_certificate(pki_node[ca_name]['certificate'])
+            if verify_certificate(cert, ca_cert):
+                verified = True
+                break
+        if not verified and name != sorted_names[-1]:
+            # Only permit top-most certificate to fail verify (e.g. signed by public CA not explicitly in chain)
+            return False
+    return True
+
 # Certificate chain
 
 def find_parent(cert, ca_certs):
@@ -357,3 +405,16 @@ def find_chain(cert, ca_certs):
             chain.append(parent)
 
     return chain
+
+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'])
+        cert2 = load_certificate(pki_node[ca_name2]['certificate'])
+
+        if verify_certificate(cert1, cert2): # cert1 is child of cert2
+            return -1
+        return 1
+
+    from functools import cmp_to_key
+    return sorted(ca_names, key=cmp_to_key(lambda cert1, cert2: ca_cmp(cert1, cert2, pki_node)))
+
diff --git a/python/vyos/util.py b/python/vyos/util.py
index 0d62fbfe9..bee5d7aec 100644
--- a/python/vyos/util.py
+++ b/python/vyos/util.py
@@ -197,7 +197,7 @@ def read_file(fname, defaultonfailure=None):
             return defaultonfailure
         raise e
 
-def write_file(fname, data, defaultonfailure=None, user=None, group=None, mode=None):
+def write_file(fname, data, defaultonfailure=None, user=None, group=None, mode=None, append=False):
     """
     Write content of data to given fname, should defaultonfailure be not None,
     it is returned on failure to read.
@@ -212,7 +212,7 @@ def write_file(fname, data, defaultonfailure=None, user=None, group=None, mode=N
     try:
         """ Write a file to string """
         bytes = 0
-        with open(fname, 'w') as f:
+        with open(fname, 'w' if not append else 'a') as f:
             bytes = f.write(data)
         chown(fname, user, group)
         chmod(fname, mode)