image: T5455: Add migration of SSH known_hosts files during image upgrade

During upgrade, the script now checks if any `known_hosts` files exist.
If so, it prompts the user to save these SSH fingerprints, and upon
confirmation, copies the files to the new image persistence directory.

Author: alexandr-san4ez

python/vyos/utils/auth.py

@@ -20,9 +20,34 @@
 
 from enum import StrEnum
 from decimal import Decimal
+from pwd import getpwall, getpwnam
 from vyos.utils.process import cmd
 
-
+# Minimum UID used when adding system users
+MIN_USER_UID: int = 1000
+# Maximim UID used when adding system users
+MAX_USER_UID: int = 59999
+# List of local user accounts that must be preserved
+SYSTEM_USER_SKIP_LIST: frozenset = {
+    'radius_user',
+    'radius_priv_user',
+    'tacacs0',
+    'tacacs1',
+    'tacacs2',
+    'tacacs3',
+    'tacacs4',
+    'tacacs5',
+    'tacacs6',
+    'tacacs7',
+    'tacacs8',
+    'tacacs9',
+    'tacacs10',
+    'tacacs11',
+    'tacacs12',
+    'tacacs13',
+    'tacacs14',
+    'tacacs15',
+}
 DEFAULT_PASSWORD: str = 'vyos'
 LOW_ENTROPY_MSG: str = 'should be at least 8 characters long;'
 WEAK_PASSWORD_MSG: str = 'The password complexity is too low - @MSG@'
@@ -119,3 +144,24 @@ def get_current_user() -> str:
     elif 'USER' in os.environ:
         current_user = os.environ['USER']
     return current_user
+
+
+def get_local_users(min_uid=MIN_USER_UID, max_uid=MAX_USER_UID) -> list:
+    """Return list of dynamically allocated users (see Debian Policy Manual)"""
+    local_users = []
+
+    for s_user in getpwall():
+        if s_user.pw_uid < min_uid:
+            continue
+        if s_user.pw_uid > max_uid:
+            continue
+        if s_user.pw_name in SYSTEM_USER_SKIP_LIST:
+            continue
+        local_users.append(s_user.pw_name)
+
+    return local_users
+
+
+def get_user_home_dir(user: str) -> str:
+    """Return user's home directory"""
+    return getpwnam(user).pw_dir

src/conf_mode/system_login.py

@@ -19,7 +19,6 @@
 from passlib.hosts import linux_context
 from psutil import users
 from pwd import getpwall
-from pwd import getpwnam
 from pwd import getpwuid
 from sys import exit
 from time import sleep
@@ -34,6 +33,7 @@
 from vyos.utils.auth import EPasswdStrength
 from vyos.utils.auth import evaluate_strength
 from vyos.utils.auth import get_current_user
+from vyos.utils.auth import MIN_USER_UID, get_local_users, get_user_home_dir
 from vyos.utils.configfs import delete_cli_node
 from vyos.utils.configfs import add_cli_node
 from vyos.utils.dict import dict_search
@@ -53,10 +53,6 @@
 tacacs_nss_config_file = "/etc/tacplus_nss.conf"
 nss_config_file = "/etc/nsswitch.conf"
 
-# Minimum UID used when adding system users
-MIN_USER_UID: int = 1000
-# Maximim UID used when adding system users
-MAX_USER_UID: int = 59999
 # LOGIN_TIMEOUT from /etc/loign.defs minus 10 sec
 MAX_RADIUS_TIMEOUT: int = 50
 # MAX_RADIUS_TIMEOUT divided by 2 sec (minimum recomended timeout)
@@ -65,25 +61,7 @@
 MAX_TACACS_COUNT: int = 8
 # Minimum USER id for TACACS users
 MIN_TACACS_UID = 900
-# List of local user accounts that must be preserved
-SYSTEM_USER_SKIP_LIST: list = ['radius_user', 'radius_priv_user', 'tacacs0', 'tacacs1',
-                              'tacacs2', 'tacacs3', 'tacacs4', 'tacacs5', 'tacacs6',
-                              'tacacs7', 'tacacs8', 'tacacs9', 'tacacs10',' tacacs11',
-                              'tacacs12', 'tacacs13', 'tacacs14', 'tacacs15']
-
-def get_local_users(min_uid=MIN_USER_UID, max_uid=MAX_USER_UID):
-    """Return list of dynamically allocated users (see Debian Policy Manual)"""
-    local_users = []
-    for s_user in getpwall():
-        if getpwnam(s_user.pw_name).pw_uid < min_uid:
-            continue
-        if getpwnam(s_user.pw_name).pw_uid > max_uid:
-            continue
-        if s_user.pw_name in SYSTEM_USER_SKIP_LIST:
-            continue
-        local_users.append(s_user.pw_name)
-
-    return local_users
+
 
 def get_shadow_password(username):
     with open('/etc/shadow') as f:
@@ -342,7 +320,7 @@ def apply(login):
                 # crazy user will choose username root or any other system user which will fail.
                 #
                 # XXX: Should we deny using root at all?
-                home_dir = getpwnam(user).pw_dir
+                home_dir = get_user_home_dir(user)
                 # always re-render SSH keys with appropriate permissions
                 render(f'{home_dir}/.ssh/authorized_keys', 'login/authorized_keys.j2',
                        user_config, permission=0o600,
@@ -397,7 +375,7 @@ def apply(login):
                 # Disable user to prevent re-login
                 call(f'usermod -s /sbin/nologin {user}')
 
-                home_dir = getpwnam(user).pw_dir
+                home_dir = get_user_home_dir(user)
                 # Remove SSH authorized keys file
                 authorized_keys_file = f'{home_dir}/.ssh/authorized_keys'
                 if os.path.exists(authorized_keys_file):

src/op_mode/image_installer.py

@@ -57,6 +57,7 @@
 from vyos.utils.file import read_file
 from vyos.utils.file import write_file
 from vyos.utils.process import cmd, run, rc_cmd
+from vyos.utils.auth import get_local_users, get_user_home_dir
 from vyos.version import get_version_data
 
 # define text messages
@@ -715,6 +716,20 @@ def copy_ssh_host_keys() -> bool:
     return False
 
 
+def copy_ssh_known_hosts() -> bool:
+    """Ask user to copy SSH `known_hosts` files
+
+    Returns:
+        bool: user's decision
+    """
+    known_hosts_files = get_known_hosts_files()
+    msg = (
+        'Would you like to save the SSH known hosts (fingerprints) '
+        'from your current configuration?'
+    )
+    return known_hosts_files and ask_yes_no(msg, default=True)
+
+
 def console_hint() -> str:
     pid = getppid() if 'SUDO_USER' in environ else getpid()
     try:
@@ -1010,6 +1025,40 @@ def install_image() -> None:
         exit(1)
 
 
+def get_known_hosts_files() -> list:
+    """Collect all existing `known_hosts` files for root and users under /home"""
+
+    files = []
+
+    # for root
+    base_files = ('/root/.ssh/known_hosts', '/etc/ssh/ssh_known_hosts')
+    for file_path in base_files:
+        root_known_hosts = Path(file_path)
+        if root_known_hosts.exists():
+            files.append(root_known_hosts)
+
+    # for each non-system user
+    for user in get_local_users():
+        home_dir = Path(get_user_home_dir(user))
+        if home_dir.exists():
+            known_hosts = home_dir / '.ssh' / 'known_hosts'
+            if known_hosts.exists():
+                files.append(known_hosts)
+
+    return files
+
+
+def migrate_known_hosts(target_dir: str):
+    """Copy `known_hosts` for root and all users to the new image directory"""
+
+    known_hosts_files = get_known_hosts_files()
+    for known_hosts_file in known_hosts_files:
+        # copy file to target directory
+        target_known_hosts = Path(f'{target_dir}{known_hosts_file}')
+        target_known_hosts.parent.mkdir(parents=True, exist_ok=True)
+        copy(known_hosts_file, target_known_hosts)
+
+
 @compat.grub_cfg_update
 def add_image(image_path: str, vrf: str = None, username: str = '',
               password: str = '', no_prompt: bool = False, force: bool = False) -> None:
@@ -1115,6 +1164,11 @@ def add_image(image_path: str, vrf: str = None, username: str = '',
             for host_key in host_keys:
                 copy(host_key, target_ssh_dir)
 
+        target_ssh_known_hosts_dir: str = f'{root_dir}/boot/{image_name}/rw'
+        if no_prompt or copy_ssh_known_hosts():
+            print('Copying SSH known_hosts files')
+            migrate_known_hosts(target_ssh_known_hosts_dir)
+
         # copy system image and kernel files
         print('Copying system image files')
         for file in Path(f'{DIR_ISO_MOUNT}/live').iterdir():

src/tests/test_utils.py

@@ -12,7 +12,12 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+import pwd
 from unittest import TestCase
+
+from vyos.utils import auth
+
+
 class TestVyOSUtils(TestCase):
     def test_key_mangling(self):
         from vyos.utils.dict import mangle_dict_keys
@@ -24,3 +29,45 @@ def test_key_mangling(self):
     def test_sysctl_read(self):
         from vyos.utils.system import sysctl_read
         self.assertEqual(sysctl_read('net.ipv4.conf.lo.forwarding'), '1')
+
+
+class TestVyOSUtilsAuth(TestCase):
+
+    def test_get_local_users_returns_existing_usernames(self):
+        # Returned users exist, skip list is excluded, and UIDs are in range
+
+        all_users = set(s_user.pw_name for s_user in pwd.getpwall())
+        local_users = auth.get_local_users()
+
+        # All returned users must really exist
+        for user in local_users:
+            self.assertIn(user, all_users)
+
+        # Nobody in the skip list
+        for skipped in auth.SYSTEM_USER_SKIP_LIST:
+            self.assertNotIn(skipped, local_users)
+
+        # All are within UID range
+        for s_user in pwd.getpwall():
+            if s_user.pw_name in local_users:
+                self.assertGreaterEqual(s_user.pw_uid, auth.MIN_USER_UID)
+                self.assertLessEqual(s_user.pw_uid, auth.MAX_USER_UID)
+
+    def test_get_user_home_dir_for_real_user(self):
+        # User's homedir is a non-empty string for a valid user
+
+        local_users = auth.get_local_users()
+        if local_users:
+            for user in local_users:
+                home_dir = auth.get_user_home_dir(user)
+                self.assertIsInstance(home_dir, str)
+                self.assertTrue(bool(home_dir))  # Should not be empty
+        else:
+            self.skipTest("No suitable non-system users found on this system")
+
+    def test_get_user_home_dir_invalid_user(self):
+        # Raises KeyError for nonexistent username
+
+        user = "__this_user_does_not_exist__"  # Test using unlikely username
+        with self.assertRaises(KeyError):
+            auth.get_user_home_dir(user)