Adding local-users-passwords-reset feature service, YANG model and its tests

Author: azmyali98

files/build_templates/init_cfg.json.j2

@@ -152,6 +152,11 @@
             "logout": ""
         }
     },
+    "LOCAL_USERS_PASSWORDS_RESET": {
+        "global": {
+            "state": "disabled"
+        }
+    },
     "SYSTEM_DEFAULTS" : {
 {%- if include_mux == "y" %}
         "mux_tunnel_egress_acl": {

files/build_templates/sonic_debian_extension.j2

@@ -620,6 +620,15 @@ sudo cp $IMAGE_CONFIGS/hostname/hostname-config.service $FILESYSTEM_ROOT_USR_LIB
 echo "hostname-config.service" | sudo tee -a $GENERATED_SERVICE_FILE
 sudo cp $IMAGE_CONFIGS/hostname/hostname-config.sh $FILESYSTEM_ROOT/usr/bin/
 
+{% if enable_local_users_passwords_reset == "y" %}
+# Copy local-users-passwords-reset configuration scripts
+sudo cp $IMAGE_CONFIGS/local-users-passwords-reset/local-users-passwords-reset.service $FILESYSTEM_ROOT_USR_LIB_SYSTEMD_SYSTEM
+echo "local-users-passwords-reset.service" | sudo tee -a $GENERATED_SERVICE_FILE
+sudo cp $IMAGE_CONFIGS/local-users-passwords-reset/local-users-passwords-reset.py $FILESYSTEM_ROOT/usr/bin/
+# Set execute permissions only
+sudo chmod 100 $FILESYSTEM_ROOT/usr/bin/
+{% endif %}
+
 # Copy banner configuration scripts
 sudo cp $IMAGE_CONFIGS/bannerconfig/banner-config.service $FILESYSTEM_ROOT_USR_LIB_SYSTEMD_SYSTEM
 echo "banner-config.service" | sudo tee -a $GENERATED_SERVICE_FILE

files/image_config/local-users-passwords-reset/local-users-passwords-reset.py

@@ -0,0 +1,62 @@
+
+import os
+import syslog
+from swsscommon.swsscommon import ConfigDBConnector, DBConnector
+from swsscommon import swsscommon
+
+
+STATE_DB = "STATE_DB"
+
+
+def get_platform_local_users_passwords_reset():
+    try:
+        from sonic_platform.local_users_passwords_reset import LocalUsersConfigurationReset
+        local_users_password_reset_class = LocalUsersConfigurationReset()
+    except ImportError:
+        syslog.syslog(syslog.LOG_WARNING, "LocalUsersConfigurationReset: sonic_platform package not installed. Unable to find platform local users passwords reset implementation")
+        raise Exception('Local users passwords reset implementation is not defined')
+
+    return local_users_password_reset_class
+
+
+class LocalUsersConfigurationResetService:
+    def __init__(self):
+        state_db_conn = DBConnector(STATE_DB, 0)
+        # Wait if the Warm/Fast boot is in progress
+        if swsscommon.RestartWaiter.isAdvancedBootInProgress(state_db_conn):
+            swsscommon.RestartWaiter.waitAdvancedBootDone()
+
+        self.config_db = ConfigDBConnector()
+        self.config_db.connect(wait_for_init=True, retry_on=True)
+        syslog.syslog(syslog.LOG_INFO, 'ConfigDB connect success')
+
+    def get_feature_state(self):
+        '''
+        Check if the feature is enabled by reading the redis table
+        '''
+        table = self.config_db.get_table(swsscommon.CFG_LOCAL_USERS_PASSWORDS_RESET)
+        if table:
+            state = table.get('global', {}).get('state')
+            return True if state == 'enabled' else False
+
+        return False
+
+    def start(self):
+        '''
+        If the feature is enabled then reset the password's using the platform
+        specific implementation
+        '''
+        local_users_password_reset = get_platform_local_users_passwords_reset()
+        feature_enabled = self.get_feature_state()
+        syslog.syslog(syslog.LOG_INFO, 'Feature is {}'.format('enabled' if feature_enabled else 'disabled'))
+        should_trigger = local_users_password_reset.should_trigger()
+        if should_trigger and feature_enabled:
+            local_users_password_reset.start()
+
+
+def main():
+    LocalUsersConfigurationResetService().start()
+
+
+if __name__ == "__main__":
+    main()

files/image_config/local-users-passwords-reset/local-users-passwords-reset.service

@@ -0,0 +1,14 @@
+[Unit]
+Description=Update Local users' passwords config based on configdb
+Requires=config-setup.service
+After=config-setup.service
+Before=systemd-logind.service sshd.service getty.target [email protected]
+
+
+[Service]
+Type=oneshot
+RemainAfterExit=no
+ExecStart=/usr/bin/local-users-passwords-reset.py
+
+[Install]
+WantedBy=sonic.target

platform/mellanox/mlnx-platform-api/sonic_platform/local_users_passwords_reset.py

@@ -0,0 +1,83 @@
+#
+# Copyright (c) 2019-2024 NVIDIA CORPORATION & AFFILIATES.
+# Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#############################################################################
+# Mellanox
+#
+# Module contains an implementation of SONiC Platform Base API and
+# provides the local users' passwords reset functionality implementation.
+#
+#############################################################################
+
+try:
+    import json
+    import subprocess
+
+    from sonic_platform_base.local_users_passwords_reset_base import LocalUsersConfigurationResetBase
+    from sonic_py_common.logger import Logger
+    from . import utils
+except ImportError as e:
+    raise ImportError (str(e) + "- required module not found")
+
+
+# Global logger class instance
+logger = Logger()
+
+
+LONG_REBOOT_PRESS_FILEPATH = '/var/run/hw-management/system/reset_long_pb'
+DEFAULT_USERS_FILEPATH = '/etc/sonic/default_users.json'
+
+
+class LocalUsersConfigurationReset(LocalUsersConfigurationResetBase):
+    def should_trigger(self):
+        '''
+        The condition for triggering passwords reset is by checking if the
+        reboot cause was a long reboot press. 
+        '''
+        try:
+            status = utils.read_int_from_file(LONG_REBOOT_PRESS_FILEPATH, raise_exception=True)
+            return True if status ==  1 else False
+        except (ValueError, IOError) as e:
+            logger.log_error(f"Failed to read long reboot press from {LONG_REBOOT_PRESS_FILEPATH} - {e}")
+            return False
+
+    @staticmethod
+    def reset_password(user, hashed_password, expire=False):
+        '''
+        This method is used to reset the user's password and expire it (optional)
+        '''
+        # Use 'chpasswd' shell command to change password
+        subprocess.call(['echo', f'{user}:{hashed_password}', '|', 'sudo', 'chpasswd', '-e'])
+        if expire:
+            # Use 'passwd' shell command to expire password
+            subprocess.call(['sudo', 'passwd', '-e', f'{user}'])
+
+    def start(self):
+        '''
+        The functionality defined is to restore original password and expire it for default local users.
+        It is done by 
+        '''
+        default_users = {}
+
+        # Fetch local users information from default_users
+        with open(DEFAULT_USERS_FILEPATH) as f:
+            default_users = json.load(f)
+
+        logger.log_info('Restoring default users\' passwords and expiring them')
+        for user in default_users.keys():
+            hashed_password = default_users.get(user, {}).get('password')
+            if hashed_password:
+                self.reset_password(user, hashed_password, expire=True)

platform/mellanox/mlnx-platform-api/tests/test_local_users_passwords_reset.py

@@ -0,0 +1,81 @@
+#
+# Copyright (c) 2020-2024 NVIDIA CORPORATION & AFFILIATES.
+# Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import os
+import pytest
+import subprocess
+import sys
+from mock import patch, mock_open
+
+test_path = os.path.dirname(os.path.abspath(__file__))
+modules_path = os.path.dirname(test_path)
+sys.path.insert(0, modules_path)
+
+from sonic_platform import utils
+from sonic_platform.local_users_passwords_reset import LocalUsersConfigurationReset
+
+
+DEFAULT_USERS_JSON_EXAMPLE_OUTPUT = '''
+{
+    "admin": {
+        "expire": "false",
+        "password": "HASHED_PASSWORD_123"
+    }
+}
+'''
+
+
+class TestLocalUsersConfigurationReset:
+    @patch('sonic_platform.utils.read_int_from_file')
+    def test_should_trigger_method(self, mock_read_int):
+        '''
+        Validate should_trigger() method
+        '''
+        local_users_reset_class = LocalUsersConfigurationReset()
+
+        mock_read_int.return_value = int(1)
+        assert local_users_reset_class.should_trigger() == True
+        mock_read_int.return_value = int(0)
+        assert local_users_reset_class.should_trigger() == False
+        mock_read_int.return_value = int(2)
+        assert local_users_reset_class.should_trigger() == False
+        mock_read_int.side_effect = ValueError()
+        assert local_users_reset_class.should_trigger() == False
+
+    @patch('subprocess.call')
+    @patch('sonic_platform.utils.read_int_from_file')
+    @patch("builtins.open", new_callable=mock_open, read_data=DEFAULT_USERS_JSON_EXAMPLE_OUTPUT)
+    def test_basic_flow_resetting_users_triggered(self, mock_read_int, mock_subproc_call):
+        '''
+        Test the basic flow of resetting local users when long button press is detected
+        '''
+        # Mock long reset button press
+        mock_read_int.return_value = int(1)
+        LocalUsersConfigurationReset().start()
+        mock_subproc_call.assert_called_with(['echo', 'admin:HASHED_PASSWORD_123', '|', 'sudo', 'chpasswd', '-e'])
+        mock_subproc_call.assert_called_with(['sudo', 'passwd', '-e', 'admin'])
+
+    @patch('subprocess.call')
+    @patch('sonic_platform.utils.read_int_from_file')
+    @patch("builtins.open", new_callable=mock_open, read_data=DEFAULT_USERS_JSON_OUTPUT)
+    def test_basic_flow_resetting_users_not_triggered(self, mock_read_int, mock_subproc_call):
+        '''
+        Test the basic flow of resetting local users when long button press is not detected
+        '''
+        # Mock long reset button NOT pressed
+        mock_read_int.return_value = int(0)
+        LocalUsersConfigurationReset().start()
+        mock_subproc_call.assert_not_called()

rules/config

@@ -51,6 +51,9 @@ DEFAULT_PASSWORD = YourPaSsWoRd
 # ENABLE_ZTP - installs Zero Touch Provisioning support.
 # ENABLE_ZTP = y
 
+# ENABLE_LOCAL_USERS_PASSWORDS_RESET - enable local users' passwords reset during init on switch
+ENABLE_LOCAL_USERS_PASSWORDS_RESET ?= y
+
 # INCLUDE_PDE - Enable platform development enviroment
 # INCLUDE_PDE = y
 # SHUTDOWN_BGP_ON_START - if set to y all bgp sessions will be in admin down state when