From 6c3b1ef2fede1e3c2b6e89060d3d645c2ba744cd Mon Sep 17 00:00:00 2001 From: Takeru Hayasaka Date: Sat, 28 Dec 2024 19:58:02 +0000 Subject: 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 --- src/conf_mode/service_ssh.py | 100 +++++++++++++++++++++++++++++++++++-------- 1 file changed, 83 insertions(+), 17 deletions(-) (limited to 'src') 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) -- cgit v1.2.3