diff --git a/freedombox/modules/ssh/privileged.py b/freedombox/modules/ssh/privileged.py index 583a7ea1f..ae1dfec35 100644 --- a/freedombox/modules/ssh/privileged.py +++ b/freedombox/modules/ssh/privileged.py @@ -2,7 +2,6 @@ """Configuration helper for SSH server.""" import grp -import os import pathlib import pwd import shutil @@ -72,10 +71,10 @@ def are_users_restricted() -> bool: return config_file.exists() -def get_user_homedir(username): +def _get_user_homedir(username: str) -> pathlib.Path: """Return the home dir of a user by looking up in password database.""" try: - return pwd.getpwnam(username).pw_dir + return pathlib.Path(pwd.getpwnam(username).pw_dir) except KeyError: raise ValueError('Username not found') @@ -83,11 +82,10 @@ def get_user_homedir(username): @privileged def get_keys(user: str) -> str: """Get SSH authorized keys.""" - path = os.path.join(get_user_homedir(user), '.ssh', 'authorized_keys') + path = _get_user_homedir(user) / '.ssh' / 'authorized_keys' try: - with open(path, 'r', encoding='utf-8') as file_handle: - return file_handle.read() - except FileNotFoundError: + return path.read_text(encoding='utf-8') + except Exception: return '' @@ -97,20 +95,18 @@ def set_keys(user: str, keys: str, auth_user: str, auth_password: secret_str): must_be_admin = user != auth_user _validate_user(auth_user, auth_password, must_be_admin=must_be_admin) - ssh_folder = os.path.join(get_user_homedir(user), '.ssh') - key_file_path = os.path.join(ssh_folder, 'authorized_keys') + ssh_folder = _get_user_homedir(user) / '.ssh' + key_file_path = ssh_folder / 'authorized_keys' action_utils.run(['mkhomedir_helper', user], check=True) - if not os.path.exists(ssh_folder): - os.makedirs(ssh_folder) + if not ssh_folder.exists(): + ssh_folder.mkdir() shutil.chown(ssh_folder, user, 'users') - with open(key_file_path, 'w', encoding='utf-8') as file_handle: - file_handle.write(keys) - + key_file_path.write_text(keys, encoding='utf-8') shutil.chown(key_file_path, user, 'users') - os.chmod(key_file_path, stat.S_IRUSR | stat.S_IWUSR) + key_file_path.chmod(stat.S_IRUSR | stat.S_IWUSR) def _load_augeas():