Security Cipher Module: Added password rotate feature and enabled Backup and Restore support for NOS Upgrades

src/sonic-py-common/sonic_py_common/security_cipher.py

@@ -164,24 +164,23 @@ def rotate_feature_passwd(self, feature_type, new_password):
             table, entry = table_info.split("|")
             db_entry = self._config_db.get_entry(table, entry)
             encrypted_passkey = db_entry.get("passkey")
-            if encrypted_passkey:
+            #Rotate only if valid passkey is present and 'key_encyrpt' flag is True
+            if encrypted_passkey and db_entry.get("key_encrypt") == 'True':
                 # Decrypt with old password
                 plain_passkey = self._decrypt_passkey(feature_type, encrypted_passkey, old_password)
                 # Re-encrypt with new password
                 new_encrypted_passkey = self._encrypt_passkey(feature_type, plain_passkey, new_password)
                 # Update DB
                 db_entry["passkey"] = new_encrypted_passkey
-                # Make sure key_encrypt should be set true
-                db_entry["key_encrypt"] = 'True'
                 self._config_db.set_entry(table, entry, db_entry)
                 syslog.syslog(syslog.LOG_INFO, "rotate_feature_passwd: Updated passkey for {}".format(table_info))
             else:
-                syslog.syslog(syslog.LOG_WARNING, "No passkey found in DB for {}".format(table_info))
+                syslog.syslog(syslog.LOG_WARNING, "Either no passkey found or key_encrypt flag is not set to True for {}".format(table_info))
 
         # Update stored password
         data[feature_type]["password"] = new_password
         self._save_registry(data)
-        syslog.syslog(syslog.LOG_INFO, f"rotate_feature_passwd: Password for {feature_type} updated.")
+        syslog.syslog(syslog.LOG_INFO, "rotate_feature_passwd: Password for {} Feature has been updated.".format(feature_type))
 
     def encrypt_passkey(self, feature_type, secret: str) -> str:
         """

src/sonic-py-common/tests/test_security_cipher.py

@@ -69,27 +69,33 @@ def test_register_table_info(self):
                 assert "RADIUS|global" in args["RADIUS"]["table_info"]
 
     def test_deregister_table_info(self):
-        test_json = {
+        # Use an in-memory registry that can be mutated
+        registry = {
             "RADIUS": {"table_info": ["RADIUS|global", "RADIUS|backup"], "password": "radius_secret"}
         }
         with mock.patch("sonic_py_common.security_cipher.ConfigDBConnector", new=ConfigDBConnector), \
-             mock.patch("os.chmod"), \
-             mock.patch("{}.open".format(BUILTINS), mock.mock_open(read_data=json.dumps(test_json))), \
-             mock.patch("os.path.exists", return_value=True):
+                mock.patch("os.chmod"), \
+                mock.patch("{}.open".format(BUILTINS), mock.mock_open()), \
+                mock.patch("os.path.exists", return_value=True):
+
             temp = master_key_mgr()
-            with mock.patch.object(temp, "_save_registry") as mock_save:
-                temp.deregister("RADIUS", "RADIUS|global")
-                args = mock_save.call_args[0][0]
-                assert "RADIUS|global" not in args["RADIUS"]["table_info"]
-                assert args["RADIUS"]["password"] == "radius_secret"
-                # Remove last and check password removed
-                temp.deregister("RADIUS", "RADIUS|backup")
-                args = mock_save.call_args[0][0]
-                assert args["RADIUS"]["table_info"] == []
-                assert args["RADIUS"]["password"] is None
+            # Patch _load_registry to always return current registry
+            temp._load_registry = mock.Mock(side_effect=lambda: registry.copy())
+            # Patch _save_registry to update our in-memory registry
+            def save_registry(data):
+                registry.clear()
+                registry.update(json.loads(json.dumps(data)))  # Deep copy
+            temp._save_registry = mock.Mock(side_effect=save_registry)
+
+            temp.deregister("RADIUS", "RADIUS|global")
+            assert registry["RADIUS"]["table_info"] == ["RADIUS|backup"]
+            assert registry["RADIUS"]["password"] == "radius_secret"
+
+            temp.deregister("RADIUS", "RADIUS|backup")
+            assert registry["RADIUS"]["table_info"] == []
+            assert registry["RADIUS"]["password"] is None
 
     def test_rotate_feature_passwd(self):
-        # Simulate DB entries and encryption/decryption
         test_json = {
             "RADIUS": {"table_info": ["RADIUS|global"], "password": "oldpw"}
         }
@@ -99,19 +105,18 @@ def test_rotate_feature_passwd(self):
              mock.patch("{}.open".format(BUILTINS), mock.mock_open(read_data=json.dumps(test_json))), \
              mock.patch("os.path.exists", return_value=True):
             temp = master_key_mgr()
+            # Patch _load_registry to return our test_json so rotate_feature_passwd sees the right state
+            temp._load_registry = mock.Mock(return_value=test_json)
             temp._config_db.get_entry = mock.Mock(return_value=db_entry.copy())
             temp._config_db.set_entry = mock.Mock()
             temp._decrypt_passkey = mock.Mock(return_value="plaintext")
             temp._encrypt_passkey = mock.Mock(return_value="NEW_ENCRYPTED")
-            with mock.patch.object(temp, "_save_registry") as mock_save:
+            with mock.patch.object(temp, "_save_registry"):
                 temp.rotate_feature_passwd("RADIUS", "newpw")
                 temp._config_db.set_entry.assert_called_once_with(
                     "RADIUS", "global",
                     {'passkey': "NEW_ENCRYPTED", "some_other_field": "keepme", "key_encrypt": 'True'}
                 )
-                # Ensure registry gets updated
-                args = mock_save.call_args[0][0]
-                assert args["RADIUS"]["password"] == "newpw"
 
     def test_encrypt_and_decrypt_passkey(self):
         # Use a known password and mock openssl subprocess

src/sonic-yang-models/yang-models/sonic-system-tacacs.yang

@@ -46,6 +46,12 @@ module sonic-system-tacacs {
         }
     }
 
+    typedef key_encrypt_type {
+        type boolean;
+        default false;
+        description "Indicates if the passkey is encrypted.";
+    }
+
     container sonic-system-tacacs {
 
         container TACPLUS_SERVER {
@@ -92,9 +98,7 @@ module sonic-system-tacacs {
                 }
 
                 leaf key_encrypt {
-                    type boolean;
-                    default false;
-                    description "Indicates if the passkey is encrypted.";
+                    type key_encrypt_type;
                 }
 
                 leaf passkey {
@@ -138,9 +142,7 @@ module sonic-system-tacacs {
                 }
 
                 leaf key_encrypt {
-                    type boolean;
-                    default false;
-                    description "Indicates if the passkey is encrypted.";
+                    type key_encrypt_type;
                 }
 
                 leaf passkey {