ssh: T6013: support SSH AuthorizedPrincipalsFile in use with trusted-user-ca-key

Thisc omplements commit e7cab89f9f81 ("T6013: Add support for configuring
TrustedUserCAKeys in SSH service with local and remote CA keys"). It introduces
a new CLI node per user to support defining the authorized principals used by
any given PKI certificate. It is now possible to associate SSH login users with
their respective principals.

Authored-by: Takeru Hayasaka <hayatake396@gmail.com>

1 files changed, 83 insertions, 17 deletions

diff --git a/src/conf_mode/service_ssh.py b/src/conf_mode/service_ssh.py
index 759f87bb2..5e6a6f771 100755
--- a/src/conf_mode/service_ssh.py
+++ b/src/conf_mode/service_ssh.py
@@ -45,6 +45,77 @@ 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'
+authorized_principals = '/etc/ssh/authorized_principals'
+
+
+def cleanup_authorized_principals_dir(valid_users: list[str]):
+    if not os.path.isdir(authorized_principals):
+        return
+
+    # Check the files (user name) under the directory and delete unnecessary ones.
+    for filename in os.listdir(authorized_principals):
+        file_path = os.path.join(authorized_principals, filename)
+        if os.path.isfile(file_path) and filename not in valid_users:
+            os.remove(file_path)
+
+    # If the directory is empty, delete it too
+    if not os.listdir(authorized_principals):
+        os.rmdir(authorized_principals)
+
+
+def handle_trusted_user_ca_key(ssh: dict):
+    if 'trusted_user_ca_key' not in ssh:
+        if os.path.exists(trusted_user_ca_key):
+            os.unlink(trusted_user_ca_key)
+
+        # remove authorized_principals directory if it exists
+        cleanup_authorized_principals_dir([])
+        return
+
+    # trusted_user_ca_key is present
+    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 'bind-user' not in ssh['trusted_user_ca_key']:
+        # remove authorized_principals directory if it exists
+        cleanup_authorized_principals_dir([])
+        return
+
+    # bind-user is present
+    configured_users = []
+    for bind_user, bind_user_config in ssh['trusted_user_ca_key']['bind-user'].items():
+        if bind_user not in ssh['login_users']:
+            raise ConfigError(f"User '{bind_user}' not found in system login users")
+
+        if 'principal' not in bind_user_config:
+            raise ConfigError(f"Principal not found for user '{bind_user}'")
+
+        principals = bind_user_config['principal']
+        if isinstance(principals, str):
+            principals = [principals]
+
+        if not os.path.isdir(authorized_principals):
+            os.makedirs(authorized_principals, exist_ok=True)
+
+        principal_file = os.path.join(authorized_principals, bind_user)
+        contents = '\n'.join(principals) + '\n'
+        write_file(principal_file, contents)
+        configured_users.append(bind_user)
+
+    # remove unnecessary files under authorized_principals directory
+    cleanup_authorized_principals_dir(configured_users)
 
 
 def get_config(config=None):
@@ -59,7 +130,15 @@ def get_config(config=None):
     ssh = conf.get_config_dict(
         base, key_mangling=('-', '_'), get_first_key=True, with_pki=True
     )
+    login_users_base = ['system', 'login', 'user']
+    login_users = conf.get_config_dict(
+        login_users_base,
+        key_mangling=('-', '_'),
+        no_tag_node_value_mangle=True,
+        get_first_key=True,
+    )
 
+    # create a list of all users, cli and users
     tmp = is_node_changed(conf, base + ['vrf'])
     if tmp:
         ssh.update({'restart_required': {}})
@@ -71,6 +150,9 @@ def get_config(config=None):
     # pass config file path - used in override template
     ssh['config_file'] = config_file
 
+    # use for trusted ca
+    ssh['login_users'] = login_users
+
     # Ignore default XML values if config doesn't exists
     # Delete key from dict
     if not conf.exists(base + ['dynamic-protection']):
@@ -119,23 +201,7 @@ 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)
-        )
-    elif os.path.exists(trusted_user_ca_key):
-        os.unlink(trusted_user_ca_key)
+    handle_trusted_user_ca_key(ssh)
 
     render(config_file, 'ssh/sshd_config.j2', ssh)