...

1 files changed, 125 insertions, 82 deletions

diff --git a/tools/ca/ca/__main__.py b/tools/ca/ca/__main__.py
index 6615da55..bfaee63a 100644
--- a/tools/ca/ca/__main__.py
+++ b/tools/ca/ca/__main__.py
@@ -32,6 +32,7 @@ import subprocess
 import json
 from leapseconddata import LeapSecondData
 from collections.abc import Iterable
+import ipaddress
 
 
 class KeyType(Enum):
@@ -208,8 +209,12 @@ def mv_bak(path):
             break
         bak_path = path.parent / f'{path.name}.bak{n}'
 
-    logger.warn('Renaming ‘%s’ to ‘%s’...', path, bak_path)
-    path.rename(bak_path)
+    try:
+        path.rename(bak_path)
+    except FileNotFoundError:
+        pass
+    else:
+        logger.warn('Renamed ‘%s’ to ‘%s’...', path, bak_path)
 
 def tai64nint(dt):
     global leapsecond_data
@@ -262,6 +267,44 @@ def write_genkey(key_type, sops, keyfile):
 
     return key
 
+def to_dn(alternative_names):
+    def go(alternative_name):
+        dn = None
+        try:
+            dn = x509.Name.from_rfc4514_string(alternative_name)
+        except ValueError:
+            pass
+
+        if dn:
+            logger.info('‘%s’ interpreted as directory name: %s', alternative_name, dn)
+            return x509.DirectoryName(dn)
+
+        addr = None
+        try:
+            addr = ipaddress.IPv4Network(alternative_name)
+        except (ipaddress.AddressValueError, ipaddress.NetmaskValueError, ValueError):
+            pass
+        try:
+            addr = ipaddress.IPv4Address(alternative_name)
+        except ipaddress.AddressValueError:
+            pass
+        try:
+            addr = ipaddress.IPv6Network(alternative_name)
+        except (ipaddress.AddressValueError, ipaddress.NetmaskValueError, ValueError):
+            pass
+        try:
+            addr = ipaddress.IPv6Address(alternative_name)
+        except ipaddress.AddressValueError:
+            pass
+
+        if addr:
+            logger.info('‘%s’ interpreted as ip address/subnet: %s', alternative_name, addr)
+            return x509.IPAddress(addr)
+
+        return x509.DNSName(alternative_name)
+
+    return map(go, alternative_names)
+
 def initca(ca_cert, ca_key, key_type, subject, clock_skew, validity, sops):
     global logger
 
@@ -273,14 +316,8 @@ def initca(ca_cert, ca_key, key_type, subject, clock_skew, validity, sops):
         if not key_type.aligned(key):
             logger.warn('Private key ‘%s’ does not align with requested type %s', ca_key, key_type)
 
-            try:
-                mv_bak(ca_key)
-            except FileNotFoundError:
-                pass
-            try:
-                mv_bak(ca_cert)
-            except FileNotFoundError:
-                pass
+            mv_bak(ca_key)
+            mv_bak(ca_cert)
 
             raise FileNotFoundError(f'Key does not align with requested type: {ca_key}')
     except FileNotFoundError:
@@ -295,9 +332,14 @@ def initca(ca_cert, ca_key, key_type, subject, clock_skew, validity, sops):
         logger.debug('Generating new certificate...')
 
         now = datetime.utcnow()
-        name = x509.Name([
-            x509.NameAttribute(NameOID.COMMON_NAME, subject.relative)
-        ])
+        name = None
+        try:
+            name = x509.Name.from_rfc4514_string(subject)
+            logger.info('‘%s’ interpreted as directory name: %s', subject, name)
+        except ValueError:
+            name = x509.Name([
+                x509.NameAttribute(NameOID.COMMON_NAME, subject)
+            ])
 
         cert = x509.CertificateBuilder().subject_name(
             name
@@ -342,24 +384,28 @@ def signcsr(ca_cert, ca_key, clock_skew, validity, subject, alternative_name, ke
         csr_bytes = csr
 
     csr = x509.load_pem_x509_csr(csr_bytes)
+    name = None
     if not subject:
-        common_name_attrs = csr.subject.get_attributes_for_oid(NameOID.COMMON_NAME)
-        if len(common_name_attrs) != 1:
-            raise InvalidParamsError('Invalid name structure in CSR')
-        subject = common_name_attrs[0].value.lower()
-        logger.warn('Using subject common name from csr: %s', subject)
-    name = x509.Name([
-        x509.NameAttribute(NameOID.COMMON_NAME, subject)
-    ])
+        name = csr.subject
+    else:
+        try:
+            name = x509.Name.from_rfc4514_string(subject)
+            logger.info('‘%s’ interpreted as directory name: %s', subject, name)
+        except ValueError:
+            name = x509.Name([
+                x509.NameAttribute(NameOID.COMMON_NAME, subject)
+            ])
 
     if not ignore_alternative_names:
         try:
             ext = csr.extensions.get_extension_for_oid(ExtensionOID.SUBJECT_ALTERNATIVE_NAME)
-            csr_alt_names = ext.value.get_values_for_type(x509.DNSName)
+            csr_alt_names = set(ext.value)
             logger.warn('Using alternative names from csr: %s', csr_alt_names)
-            alternative_name = list(set(alternative_name) | set(csr_alt_names))
+            alternative_name = set(to_dn(alternative_name)) | csr_alt_names
         except ExtensionNotFound:
             pass
+    else:
+        alternative_name = to_dn(alternative_name)
 
     ca_key = load_key(ca_key)
     with open(ca_cert, 'rb') as fh:
@@ -397,9 +443,7 @@ def signcsr(ca_cert, ca_key, clock_skew, validity, subject, alternative_name, ke
 
     if alternative_name:
         cert = cert.add_extension(
-            x509.SubjectAlternativeName(
-                list(map(x509.DNSName, alternative_name))
-            ),
+            x509.SubjectAlternativeName(alternative_name),
             False
         )
 
@@ -407,10 +451,7 @@ def signcsr(ca_cert, ca_key, clock_skew, validity, subject, alternative_name, ke
 
     output = output.with_suffix('.crt')
 
-    try:
-        mv_bak(output)
-    except FileNotFoundError:
-        pass
+    mv_bak(output)
     with umask(0o0133), atomic_write(output, overwrite=False, mode='wb') as cf:
         logger.info('Writing new certificate to ‘%s’...', output)
         cf.write(cert.public_bytes(serialization.Encoding.PEM))
@@ -429,26 +470,28 @@ def new_client(ca_cert, ca_key, key_type, clock_skew, validity, subject, alterna
         if not key_type.aligned(key):
             logger.warn('Private key ‘%s’ does not align with requested type %s', key_file, key_type)
 
-            try:
-                mv_bak(key_file)
-            except FileNotFoundError:
-                pass
-            try:
-                mv_bak(cert_file)
-            except FileNotFoundError:
-                pass
+            mv_bak(key_file)
+            mv_bak(cert_file)
 
             raise FileNotFoundError(f'Key does not align with requested type: {key_file}')
     except FileNotFoundError:
         key = write_genkey(key_type, sops, key_file)
 
-    csr = x509.CertificateSigningRequestBuilder().subject_name(x509.Name([
-        x509.NameAttribute(NameOID.COMMON_NAME, subject)
-    ]))
+    name = None
+    try:
+        name = x509.Name.from_rfc4514_string(subject)
+        logger.info('‘%s’ interpreted as directory name: %s', subject, name)
+    except ValueError:
+        name = x509.Name([
+            x509.NameAttribute(NameOID.COMMON_NAME, subject)
+        ])
+
+    csr = x509.CertificateSigningRequestBuilder().subject_name(name)
+
     if alternative_name:
         csr = csr.add_extension(
             x509.SubjectAlternativeName(
-                list(map(x509.DNSName, alternative_name))
+                to_dn(alternative_name)
             ),
             False
         )
@@ -553,54 +596,54 @@ def main():
 
 
     parser = argparse.ArgumentParser(prog='ca', formatter_class=argparse.ArgumentDefaultsHelpFormatter)
-    parser.add_argument('--verbosity', dest='log_level', action='append', type=int)
-    parser.add_argument('--verbose', '-v', dest='log_level', action='append_const', const=1)
-    parser.add_argument('--quiet', '-q', dest='log_level', action='append_const', const=-1)
+    parser.add_argument('--verbosity', dest='log_level', action='append', type=int, help='Numeric verbosity')
+    parser.add_argument('--verbose', '-v', dest='log_level', action='append_const', const=1, help='Increase verbosity')
+    parser.add_argument('--quiet', '-q', dest='log_level', action='append_const', const=-1, help='Decrease verbosity')
     subparsers = parser.add_subparsers(help='Subcommands', required=True)
 
-    subparser = subparsers.add_parser('init', aliases=['initca', 'init-ca', 'ca'], formatter_class=argparse.ArgumentDefaultsHelpFormatter)
-    subparser.add_argument('--ca-cert', type=Path, default=Path('ca.crt'))
-    subparser.add_argument('--ca-key', type=Path, default=Path('ca.key'))
-    subparser.add_argument('--key-type', type=KeyType.from_string, choices=list(KeyType), default=KeyType.ED448.value)
-    subparser.add_argument('--clock-skew', metavar='DURATION', type=duration, default=timedelta(minutes=5))
-    subparser.add_argument('--validity', metavar='DURATION', type=duration, default=timedelta(days=ceil(365.2425*10)))
-    subparser.add_argument('--sops', '--no-sops', action=BooleanAction, default=True)
-    subparser.add_argument('--subject', metavar='FQDN', type=ValidFQDN, required=True)
+    subparser = subparsers.add_parser('init', aliases=['initca', 'init-ca', 'ca'], formatter_class=argparse.ArgumentDefaultsHelpFormatter, description="Generate a new selfsigned CA certificate and associated private key\n\nPrivate key is only generated if it does not yet exist")
+    subparser.add_argument('--ca-cert', type=Path, default=Path('ca.crt'), help='Path to file containing CA certificate')
+    subparser.add_argument('--ca-key', type=Path, default=Path('ca.key'), help='Path to file containing CA private key')
+    subparser.add_argument('--key-type', type=KeyType.from_string, choices=list(KeyType), default=KeyType.ED448.value, help='Type of private key to generate')
+    subparser.add_argument('--clock-skew', metavar='DURATION', type=duration, default=timedelta(minutes=5), help='How far to shift begin of validity into the past')
+    subparser.add_argument('--validity', metavar='DURATION', type=duration, default=timedelta(days=ceil(365.2425*10)), help='How far to shift end of validity into the future')
+    subparser.add_argument('--sops', '--no-sops', action=BooleanAction, default=True, help='Encrypt private key using SOPS')
+    subparser.add_argument('--subject', metavar='DN', type=str, required=True, help='Subject name')
     subparser.set_defaults(cmd=initca)
 
-    subparser = subparsers.add_parser('sign', aliases=['signcsr', 'sign-csr'], formatter_class=argparse.ArgumentDefaultsHelpFormatter)
-    subparser.add_argument('--ca-cert', type=Path, default=Path('ca.crt'))
-    subparser.add_argument('--ca-key', type=Path, default=Path('ca.key'))
-    subparser.add_argument('--clock-skew', metavar='DURATION', type=duration, default=timedelta(minutes=5))
-    subparser.add_argument('--validity', metavar='DURATION', type=duration, default=timedelta(days=ceil(365.2425*10)))
-    subparser.add_argument('--subject', metavar='CN', type=str, required=False)
-    subparser.add_argument('--ignore-alternative-names', '--no-ignore-alternative-names', action=BooleanAction, default=True)
-    subparser.add_argument('--key-usage', metavar='KEY_USAGE', type=SupportedKeyUsage, action=ExtendAction, default=[SupportedKeyUsage.CLIENT_AUTH])
-    subparser.add_argument('--alternative-name', metavar='CN', type=str, action='append')
-    subparser.add_argument('--output', type=Path, required=True)
-    subparser.add_argument('csr', metavar='FILE', type=argparse.FileType(mode='rb'))
+    subparser = subparsers.add_parser('sign', aliases=['signcsr', 'sign-csr'], formatter_class=argparse.ArgumentDefaultsHelpFormatter, description='Sign an existing CSR')
+    subparser.add_argument('--ca-cert', type=Path, default=Path('ca.crt'), help='Path to file containing CA certificate')
+    subparser.add_argument('--ca-key', type=Path, default=Path('ca.key'), help='Path to file containing CA private key')
+    subparser.add_argument('--clock-skew', metavar='DURATION', type=duration, default=timedelta(minutes=5), help='How far to shift begin of validity into the past')
+    subparser.add_argument('--validity', metavar='DURATION', type=duration, default=timedelta(days=ceil(365.2425*10)), help='How far to shift end of validity into the future')
+    subparser.add_argument('--subject', metavar='DN', type=str, required=False, help='Override subject name')
+    subparser.add_argument('--ignore-alternative-names', '--no-ignore-alternative-names', action=BooleanAction, default=True, help='Ignore subject alternative names provided in CSR')
+    subparser.add_argument('--key-usage', metavar='KEY_USAGE', type=SupportedKeyUsage, action=ExtendAction, default=[SupportedKeyUsage.CLIENT_AUTH], help='Allowed key usages')
+    subparser.add_argument('--alternative-name', metavar='CN', type=str, action='append', help='Subject alternative names')
+    subparser.add_argument('--output', type=Path, required=True, help='Output path')
+    subparser.add_argument('csr', metavar='FILE', type=argparse.FileType(mode='rb'), help='Path to file containing CSR')
     subparser.set_defaults(cmd=signcsr)
 
-    subparser = subparsers.add_parser('new-client', aliases=['new', 'new-client', 'client'], formatter_class=argparse.ArgumentDefaultsHelpFormatter)
-    subparser.add_argument('--ca-cert', type=Path, default=Path('ca.crt'))
-    subparser.add_argument('--ca-key', type=Path, default=Path('ca.key'))
-    subparser.add_argument('--key-type', type=KeyType.from_string, choices=list(KeyType), default=KeyType.ED25519.value)
-    subparser.add_argument('--clock-skew', metavar='DURATION', type=duration, default=timedelta(minutes=5))
-    subparser.add_argument('--validity', metavar='DURATION', type=duration, default=timedelta(days=ceil(365.2425*10)))
-    subparser.add_argument('--sops', '--no-sops', action=BooleanAction, default=True)
-    subparser.add_argument('--subject', metavar='CN', type=str, required=True)
-    subparser.add_argument('--key-usage', metavar='KEY_USAGE', type=SupportedKeyUsage, action=ExtendAction, default=[SupportedKeyUsage.CLIENT_AUTH])
-    subparser.add_argument('--alternative-name', metavar='CN', type=str, action='append')
-    subparser.add_argument('--output', type=Path, required=True)
+    subparser = subparsers.add_parser('new-client', aliases=['new', 'new-client', 'client'], formatter_class=argparse.ArgumentDefaultsHelpFormatter, description='Generate a new CSR and sign it immediately')
+    subparser.add_argument('--ca-cert', type=Path, default=Path('ca.crt'), help='Path to file containing CA certificate')
+    subparser.add_argument('--ca-key', type=Path, default=Path('ca.key'), help='Path to file containing CA private key')
+    subparser.add_argument('--key-type', type=KeyType.from_string, choices=list(KeyType), default=KeyType.ED25519.value, help='Type of private key to generate')
+    subparser.add_argument('--clock-skew', metavar='DURATION', type=duration, default=timedelta(minutes=5), help='How far to shift begin of validity into the past')
+    subparser.add_argument('--validity', metavar='DURATION', type=duration, default=timedelta(days=ceil(365.2425*10)), help='How far to shift end of validity into the future')
+    subparser.add_argument('--sops', '--no-sops', action=BooleanAction, default=True, help='Encrypt private key using SOPS')
+    subparser.add_argument('--subject', metavar='DN', type=str, required=True, help='Subject name')
+    subparser.add_argument('--key-usage', metavar='KEY_USAGE', type=SupportedKeyUsage, action=ExtendAction, default=[SupportedKeyUsage.CLIENT_AUTH], help='Allowed key usages')
+    subparser.add_argument('--alternative-name', metavar='CN', type=str, action='append', help='Subject alternative names')
+    subparser.add_argument('--output', type=Path, required=True, help='Output path')
     subparser.set_defaults(cmd=new_client)
 
-    subparser = subparsers.add_parser('pkcs12', aliases=['p12', 'pfx'], formatter_class=argparse.ArgumentDefaultsHelpFormatter)
-    subparser.add_argument('--random-password', '--no-random-password', action=BooleanAction, default=True)
-    subparser.add_argument('--random-password-length', type=int, default=12)
-    subparser.add_argument('--weak-encryption', '--no-weak-encryption', action=BooleanAction, default=False)
-    subparser.add_argument('--temporary-output', '--no-temporary-output', action=BooleanAction, default=True)
-    subparser.add_argument('--output', type=Path)
-    subparser.add_argument('filename', metavar='BASENAME', type=Path)
+    subparser = subparsers.add_parser('pkcs12', aliases=['p12', 'pfx'], formatter_class=argparse.ArgumentDefaultsHelpFormatter, description='Convert existing certificate and private key to PKCS#12 format')
+    subparser.add_argument('--random-password', '--no-random-password', action=BooleanAction, default=True, help='Encrypt PKCS#12 file with random passphrase -- otherwise prompt for one')
+    subparser.add_argument('--random-password-length', type=int, default=12, help='Number of words in random passphrase')
+    subparser.add_argument('--weak-encryption', '--no-weak-encryption', action=BooleanAction, default=False, help='Use weak, but more compatible, encryption')
+    subparser.add_argument('--temporary-output', '--no-temporary-output', action=BooleanAction, default=True, help='If output path is not given, generate output file in temporary directory')
+    subparser.add_argument('--output', type=Path, help='Output path')
+    subparser.add_argument('filename', metavar='BASENAME', type=Path, help='Input path')
     subparser.set_defaults(cmd=to_pkcs12)
 
     args = parser.parse_args()