ssh: T6013: rename trusted-user-ca-key -> truster-user-ca

The current implementation for SSH CA based authentication uses "set service
ssh trusted-user-ca-key ca-certificate <foo>" to define an X.509 certificate
from "set pki ca <foo> ..." - fun fact, native OpenSSH does not support X.509
certificates and only runs with OpenSSH ssh-keygen generated RSA or EC keys.

This commit changes the bahavior to support antive certificates generated using
ssh-keygen and loaded to our PKI tree. As the previous implementation
did not work at all, no migrations cript is used.

4 files changed, 48 insertions, 39 deletions

diff --git a/src/conf_mode/pki.py b/src/conf_mode/pki.py
index 14fe86d56..7d01b6642 100755
--- a/src/conf_mode/pki.py
+++ b/src/conf_mode/pki.py
@@ -64,7 +64,7 @@ sync_search = [
         'path': ['service', 'https'],
     },
     {
-        'keys': ['ca_certificate'],
+        'keys': ['key'],
         'path': ['service', 'ssh'],
     },
     {
@@ -418,7 +418,8 @@ def verify(pki):
             if 'country' in default_values:
                 country = default_values['country']
                 if len(country) != 2 or not country.isalpha():
-                    raise ConfigError(f'Invalid default country value. Value must be 2 alpha characters.')
+                    raise ConfigError('Invalid default country value. '\
+                                      'Value must be 2 alpha characters.')
 
     if 'changed' in pki:
         # if the list is getting longer, we can move to a dict() and also embed the
diff --git a/src/conf_mode/service_ssh.py b/src/conf_mode/service_ssh.py
index f3c76508b..3d38d940a 100755
--- a/src/conf_mode/service_ssh.py
+++ b/src/conf_mode/service_ssh.py
@@ -23,14 +23,14 @@ from syslog import LOG_INFO
 from vyos.config import Config
 from vyos.configdict import is_node_changed
 from vyos.configverify import verify_vrf
-from vyos.configverify import verify_pki_ca_certificate
+from vyos.configverify import verify_pki_openssh_key
+from vyos.defaults import config_files
 from vyos.utils.process import call
 from vyos.template import render
 from vyos import ConfigError
 from vyos import airbag
-from vyos.pki import find_chain
-from vyos.pki import encode_certificate
-from vyos.pki import load_certificate
+from vyos.pki import encode_public_key
+from vyos.pki import load_openssh_public_key
 from vyos.utils.dict import dict_search_recursive
 from vyos.utils.file import write_file
 
@@ -45,7 +45,7 @@ key_rsa = '/etc/ssh/ssh_host_rsa_key'
 key_dsa = '/etc/ssh/ssh_host_dsa_key'
 key_ed25519 = '/etc/ssh/ssh_host_ed25519_key'
 
-trusted_user_ca_key = '/etc/ssh/trusted_user_ca_key'
+trusted_user_ca = config_files['sshd_user_ca']
 
 def get_config(config=None):
     if config:
@@ -79,9 +79,9 @@ def get_config(config=None):
                                 get_first_key=True)
 
     for value, _ in dict_search_recursive(tmp, 'principal'):
-        # Only enable principal handling if SSH trusted-user-ca-key is set
-        if 'trusted_user_ca_key' in ssh:
-            ssh['trusted_user_ca_key'].update({'has_principals': {}})
+        # Only enable principal handling if SSH trusted-user-ca is set
+        if 'trusted_user_ca' in ssh:
+            ssh['has_principals'] = {}
         # We do only need to execute this code path once as we need to know
         # if any one of the local users has a principal set or not - this
         # accounts for the entire system.
@@ -97,16 +97,8 @@ def verify(ssh):
     if 'rekey' in ssh and 'data' not in ssh['rekey']:
         raise ConfigError('Rekey data is required!')
 
-    if 'trusted_user_ca_key' in ssh:
-        if 'ca_certificate' not in ssh['trusted_user_ca_key']:
-            raise ConfigError('CA certificate is mandatory when using ' \
-                              'trusted-user-ca-key')
-
-        ca_key_name = ssh['trusted_user_ca_key']['ca_certificate']
-        verify_pki_ca_certificate(ssh, ca_key_name)
-        pki_ca_cert = ssh['pki']['ca'][ca_key_name]
-        if 'certificate' not in pki_ca_cert or not pki_ca_cert['certificate']:
-            raise ConfigError(f"CA certificate '{ca_key_name}' is not valid or missing")
+    if 'trusted_user_ca' in ssh:
+        verify_pki_openssh_key(ssh, ssh['trusted_user_ca'])
 
     verify_vrf(ssh)
     return None
@@ -131,23 +123,17 @@ def generate(ssh):
         syslog(LOG_INFO, 'SSH ed25519 host key not found, generating new key!')
         call(f'ssh-keygen -q -N "" -t ed25519 -f {key_ed25519}')
 
-    if 'trusted_user_ca_key' in ssh:
-        ca_key_name = ssh['trusted_user_ca_key']['ca_certificate']
-        pki_ca_cert = ssh['pki']['ca'][ca_key_name]
-
-        loaded_ca_cert = load_certificate(pki_ca_cert['certificate'])
-        loaded_ca_certs = {
-            load_certificate(c['certificate'])
-            for c in ssh['pki']['ca'].values()
-            if 'certificate' in c
-        }
-
-        ca_full_chain = find_chain(loaded_ca_cert, loaded_ca_certs)
-        write_file(trusted_user_ca_key,
-                   '\n'.join(encode_certificate(c) for c in ca_full_chain))
+    if 'trusted_user_ca' in ssh:
+        key_name = ssh['trusted_user_ca']
+        openssh_cert = ssh['pki']['openssh'][key_name]
+        loaded_ca_cert = load_openssh_public_key(openssh_cert['public']['key'],
+                                                 openssh_cert['public']['type'])
+        tmp = encode_public_key(loaded_ca_cert, encoding='OpenSSH',
+                                key_format='OpenSSH')
+        write_file(trusted_user_ca, tmp, trailing_newline=True)
     else:
-        if os.path.exists(trusted_user_ca_key):
-            os.unlink(trusted_user_ca_key)
+        if os.path.exists(trusted_user_ca):
+            os.unlink(trusted_user_ca)
 
     render(config_file, 'ssh/sshd_config.j2', ssh)
 
diff --git a/src/conf_mode/system_login.py b/src/conf_mode/system_login.py
index 481fdd16e..22b6fcc98 100755
--- a/src/conf_mode/system_login.py
+++ b/src/conf_mode/system_login.py
@@ -375,14 +375,15 @@ def apply(login):
                     chown(home_dir, user=user, recursive=True)
 
             # Generate 2FA/MFA One-Time-Pad configuration
+            google_auth_file = f'{home_dir}/.google_authenticator'
             if dict_search('authentication.otp.key', user_config):
                 enable_otp = True
-                render(f'{home_dir}/.google_authenticator', 'login/pam_otp_ga.conf.j2',
+                render(google_auth_file, 'login/pam_otp_ga.conf.j2',
                        user_config, permission=0o400, user=user, group='users')
             else:
                 # delete configuration as it's not enabled for the user
-                if os.path.exists(f'{home_dir}/.google_authenticator'):
-                    os.remove(f'{home_dir}/.google_authenticator')
+                if os.path.exists(google_auth_file):
+                    os.unlink(google_auth_file)
 
             # Lock/Unlock local user account
             lock_unlock = '--unlock'
@@ -396,6 +397,22 @@ def apply(login):
                 # Disable user to prevent re-login
                 call(f'usermod -s /sbin/nologin {user}')
 
+                home_dir = getpwnam(user).pw_dir
+                # Remove SSH authorized keys file
+                authorized_keys_file = f'{home_dir}/.ssh/authorized_keys'
+                if os.path.exists(authorized_keys_file):
+                    os.unlink(authorized_keys_file)
+
+                # Remove SSH authorized principals file
+                principals_file = f'{home_dir}/.ssh/authorized_principals'
+                if os.path.exists(principals_file):
+                    os.unlink(principals_file)
+
+                # Remove Google Authenticator file
+                google_auth_file = f'{home_dir}/.google_authenticator'
+                if os.path.exists(google_auth_file):
+                    os.unlink(google_auth_file)
+
                 # Logout user if he is still logged in
                 if user in list(set([tmp[0] for tmp in users()])):
                     print(f'{user} is logged in, forcing logout!')
diff --git a/src/tests/test_template.py b/src/tests/test_template.py
index 7cae867a0..4660c0038 100644
--- a/src/tests/test_template.py
+++ b/src/tests/test_template.py
@@ -192,10 +192,15 @@ class TestVyOSTemplate(TestCase):
             self.assertIn(IKEv2_DEFAULT, ','.join(ciphers))
 
     def test_get_default_port(self):
+        from vyos.defaults import config_files
         from vyos.defaults import internal_ports
 
         with self.assertRaises(RuntimeError):
+            vyos.template.get_default_config_file('UNKNOWN')
+        with self.assertRaises(RuntimeError):
             vyos.template.get_default_port('UNKNOWN')
 
+        self.assertEqual(vyos.template.get_default_config_file('sshd_user_ca'),
+                         config_files['sshd_user_ca'])
         self.assertEqual(vyos.template.get_default_port('certbot_haproxy'),
                          internal_ports['certbot_haproxy'])