T6013: Add support for configuring TrustedUserCAKeys in SSH service w…

…ith local and remote CA keys

data/templates/ssh/sshd_config.j2

@@ -110,3 +110,7 @@ ClientAliveInterval {{ client_keepalive_interval }}
 {% if rekey.data is vyos_defined  %}
 RekeyLimit {{ rekey.data }}M {{ rekey.time + 'M' if rekey.time is vyos_defined }}
 {% endif %}
+
+{% if trusted_ca is vyos_defined %}
+TrustedUserCAKeys /etc/ssh/trusted_user_ca_key
+{% endif %}

interface-definitions/service_ssh.xml.in

@@ -275,6 +275,14 @@
               </constraint>
             </properties>
           </leafNode>
+          <node name="trusted-user-ca-key">
+            <properties>
+              <help>Trusted user CA key</help>
+            </properties>
+            <children>
+              #include <include/pki/ca-certificate.xml.i>
+            </children>
+          </node>
           #include <include/vrf-multi.xml.i>
         </children>
       </node>

smoketest/scripts/cli/test_service_ssh.py

@@ -33,16 +33,20 @@
 PROCESS_NAME = 'sshd'
 SSHD_CONF = '/run/sshd/sshd_config'
 base_path = ['service', 'ssh']
+pki_path = ['pki']
 
 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'
+
 
 def get_config_value(key):
     tmp = read_file(SSHD_CONF)
     tmp = re.findall(f'\n?{key}\s+(.*)', tmp)
     return tmp
 
+
 class TestServiceSSH(VyOSUnitTestSHIM.TestCase):
     @classmethod
     def setUpClass(cls):
@@ -98,27 +102,27 @@ def test_ssh_single_listen_address(self):
 
         # Check configured port
         port = get_config_value('Port')[0]
-        self.assertTrue("1234" in port)
+        self.assertTrue('1234' in port)
 
         # Check DNS usage
         dns = get_config_value('UseDNS')[0]
-        self.assertTrue("no" in dns)
+        self.assertTrue('no' in dns)
 
         # Check PasswordAuthentication
         pwd = get_config_value('PasswordAuthentication')[0]
-        self.assertTrue("no" in pwd)
+        self.assertTrue('no' in pwd)
 
         # Check loglevel
         loglevel = get_config_value('LogLevel')[0]
-        self.assertTrue("VERBOSE" in loglevel)
+        self.assertTrue('VERBOSE' in loglevel)
 
         # Check listen address
         address = get_config_value('ListenAddress')[0]
-        self.assertTrue("127.0.0.1" in address)
+        self.assertTrue('127.0.0.1' in address)
 
         # Check keepalive
         keepalive = get_config_value('ClientAliveInterval')[0]
-        self.assertTrue("100" in keepalive)
+        self.assertTrue('100' in keepalive)
 
     def test_ssh_multiple_listen_addresses(self):
         # Check if SSH service can be configured and runs with multiple
@@ -197,7 +201,17 @@ def test_ssh_login(self):
         test_command = 'uname -a'
 
         self.cli_set(base_path)
-        self.cli_set(['system', 'login', 'user', test_user, 'authentication', 'plaintext-password', test_pass])
+        self.cli_set(
+            [
+                'system',
+                'login',
+                'user',
+                test_user,
+                'authentication',
+                'plaintext-password',
+                test_pass,
+            ]
+        )
 
         # commit changes
         self.cli_commit()
@@ -210,7 +224,9 @@ def test_ssh_login(self):
 
         # Login with invalid credentials
         with self.assertRaises(paramiko.ssh_exception.AuthenticationException):
-            output, error = self.ssh_send_cmd(test_command, 'invalid_user', 'invalid_password')
+            output, error = self.ssh_send_cmd(
+                test_command, 'invalid_user', 'invalid_password'
+            )
 
         self.cli_delete(['system', 'login', 'user', test_user])
         self.cli_commit()
@@ -250,7 +266,7 @@ def test_ssh_dynamic_protection(self):
         sshguard_lines = [
             f'THRESHOLD={threshold}',
             f'BLOCK_TIME={block_time}',
-            f'DETECTION_TIME={detect_time}'
+            f'DETECTION_TIME={detect_time}',
         ]
 
         tmp_sshguard_conf = read_file(SSHGUARD_CONFIG)
@@ -268,12 +284,16 @@ def test_ssh_dynamic_protection(self):
 
         self.assertFalse(process_named_running(SSHGUARD_PROCESS))
 
-
     # Network Device Collaborative Protection Profile
     def test_ssh_ndcpp(self):
         ciphers = ['aes128-cbc', 'aes128-ctr', 'aes256-cbc', 'aes256-ctr']
         host_key_algs = ['sk-ssh-ed25519@openssh.com', 'ssh-rsa', 'ssh-ed25519']
-        kexes = ['diffie-hellman-group14-sha1', 'ecdh-sha2-nistp256', 'ecdh-sha2-nistp384', 'ecdh-sha2-nistp521']
+        kexes = [
+            'diffie-hellman-group14-sha1',
+            'ecdh-sha2-nistp256',
+            'ecdh-sha2-nistp384',
+            'ecdh-sha2-nistp521',
+        ]
         macs = ['hmac-sha1', 'hmac-sha2-256', 'hmac-sha2-512']
         rekey_time = '60'
         rekey_data = '1024'
@@ -293,22 +313,29 @@ def test_ssh_ndcpp(self):
         # commit changes
         self.cli_commit()
 
-        ssh_lines = ['Ciphers aes128-cbc,aes128-ctr,aes256-cbc,aes256-ctr',
-                     'HostKeyAlgorithms sk-ssh-ed25519@openssh.com,ssh-rsa,ssh-ed25519',
-                     'MACs hmac-sha1,hmac-sha2-256,hmac-sha2-512',
-                     'KexAlgorithms diffie-hellman-group14-sha1,ecdh-sha2-nistp256,ecdh-sha2-nistp384,ecdh-sha2-nistp521',
-                     'RekeyLimit 1024M 60M'
-                     ]
+        ssh_lines = [
+            'Ciphers aes128-cbc,aes128-ctr,aes256-cbc,aes256-ctr',
+            'HostKeyAlgorithms sk-ssh-ed25519@openssh.com,ssh-rsa,ssh-ed25519',
+            'MACs hmac-sha1,hmac-sha2-256,hmac-sha2-512',
+            'KexAlgorithms diffie-hellman-group14-sha1,ecdh-sha2-nistp256,ecdh-sha2-nistp384,ecdh-sha2-nistp521',
+            'RekeyLimit 1024M 60M',
+        ]
         tmp_sshd_conf = read_file(SSHD_CONF)
 
         for line in ssh_lines:
             self.assertIn(line, tmp_sshd_conf)
 
     def test_ssh_pubkey_accepted_algorithm(self):
-        algs = ['ssh-ed25519', 'ecdsa-sha2-nistp256', 'ecdsa-sha2-nistp384',
-                'ecdsa-sha2-nistp521', 'ssh-dss', 'ssh-rsa', 'rsa-sha2-256',
-                'rsa-sha2-512'
-                ]
+        algs = [
+            'ssh-ed25519',
+            'ecdsa-sha2-nistp256',
+            'ecdsa-sha2-nistp384',
+            'ecdsa-sha2-nistp521',
+            'ssh-dss',
+            'ssh-rsa',
+            'rsa-sha2-256',
+            'rsa-sha2-512',
+        ]
 
         expected = 'PubkeyAcceptedAlgorithms '
         for alg in algs:
@@ -320,6 +347,38 @@ def test_ssh_pubkey_accepted_algorithm(self):
         tmp_sshd_conf = read_file(SSHD_CONF)
         self.assertIn(expected, tmp_sshd_conf)
 
+    def test_ssh_ca_certificate_single(self):
+        ca_cert_name = 'test_ca'
+        ca_key_data = 'dummy-ca-key-data'
+
+        # set pki ca certificate <ca_cert_name> certificate dummy-<ca_cert_name>-certificate <ca_key_data>
+        # set service ssh trusted-user-ca-key ca-certificate <ca_cert_name>
+        self.cli_set(
+            pki_path
+            + ['ca', ca_cert_name, 'certificate', f'dummy-{ca_cert_name}-certificate'],
+            ca_key_data,
+        )
+        self.cli_set(
+            base_path + ['trusted-user-ca-key', 'ca-certificate', ca_cert_name]
+        )
+
+        self.cli_commit()
+
+        trusted_user_ca_key_config = get_config_value('TrustedUserCAKeys')
+        self.assertIn(trusted_user_ca_key, trusted_user_ca_key_config)
+
+        with open(trusted_user_ca_key, 'r') as file:
+            ca_key_contents = file.read()
+        self.assertIn(f'dummy-{ca_cert_name}-certificate', ca_key_contents)
+
+        self.cli_delete(base_path + ['trusted-user-ca-key'])
+        self.cli_delete(['pki', 'ca', ca_cert_name])
+        self.cli_commit()
+
+        # Verify the CA key is removed
+        trusted_user_ca_key_config = get_config_value('TrustedUserCAKeys')
+        self.assertNotIn(trusted_user_ca_key, trusted_user_ca_key_config)
+
 
 if __name__ == '__main__':
     unittest.main(verbosity=2)

src/conf_mode/service_ssh.py

@@ -23,10 +23,16 @@
 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.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.utils.file import write_file
+
 airbag.enable()
 
 config_file = r'/run/sshd/sshd_config'
@@ -38,6 +44,9 @@
 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'
+
+
 def get_config(config=None):
     if config:
         conf = config
@@ -47,10 +56,13 @@ def get_config(config=None):
     if not conf.exists(base):
         return None
 
-    ssh = conf.get_config_dict(base, key_mangling=('-', '_'), get_first_key=True)
+    ssh = conf.get_config_dict(
+        base, key_mangling=('-', '_'), get_first_key=True, with_pki=True
+    )
 
     tmp = is_node_changed(conf, base + ['vrf'])
-    if tmp: ssh.update({'restart_required': {}})
+    if tmp:
+        ssh.update({'restart_required': {}})
 
     # We have gathered the dict representation of the CLI, but there are default
     # options which we need to update into the dictionary retrived.
@@ -62,20 +74,32 @@ def get_config(config=None):
     # Ignore default XML values if config doesn't exists
     # Delete key from dict
     if not conf.exists(base + ['dynamic-protection']):
-         del ssh['dynamic_protection']
+        del ssh['dynamic_protection']
 
     return ssh
 
+
 def verify(ssh):
     if not ssh:
         return None
 
     if 'rekey' in ssh and 'data' not in ssh['rekey']:
-        raise ConfigError(f'Rekey data is required!')
+        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 required for TrustedUserCAKey')
+
+        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")
 
     verify_vrf(ssh)
     return None
 
+
 def generate(ssh):
     if not ssh:
         if os.path.isfile(config_file):
@@ -95,6 +119,22 @@ 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)
+        )
+
     render(config_file, 'ssh/sshd_config.j2', ssh)
 
     if 'dynamic_protection' in ssh:
@@ -103,12 +143,13 @@ def generate(ssh):
 
     return None
 
+
 def apply(ssh):
     systemd_service_ssh = 'ssh.service'
     systemd_service_sshguard = 'sshguard.service'
     if not ssh:
         # SSH access is removed in the commit
-        call(f'systemctl stop ssh@*.service')
+        call('systemctl stop ssh@*.service')
         call(f'systemctl stop {systemd_service_sshguard}')
         return None
 
@@ -122,13 +163,14 @@ def apply(ssh):
     if 'restart_required' in ssh:
         # this is only true if something for the VRFs changed, thus we
         # stop all VRF services and only restart then new ones
-        call(f'systemctl stop ssh@*.service')
+        call('systemctl stop ssh@*.service')
         systemd_action = 'restart'
 
     for vrf in ssh['vrf']:
         call(f'systemctl {systemd_action} ssh@{vrf}.service')
     return None
 
+
 if __name__ == '__main__':
     try:
         c = get_config()