From a7708da78916c1fa8bde005be021b8a2e5ffaf98 Mon Sep 17 00:00:00 2001 From: nmoray Date: Thu, 16 Nov 2023 06:35:50 -0800 Subject: [PATCH 01/26] Added a common encrypt/decrypt module w.r.t. HLD:TACACSPLUS_PASSKEY_ENCRYPTION --- .../sonic_py_common/security_cipher.py | 120 ++++++++++++++++++ 1 file changed, 120 insertions(+) create mode 100644 src/sonic-py-common/sonic_py_common/security_cipher.py diff --git a/src/sonic-py-common/sonic_py_common/security_cipher.py b/src/sonic-py-common/sonic_py_common/security_cipher.py new file mode 100644 index 000000000000..13b8b45b7546 --- /dev/null +++ b/src/sonic-py-common/sonic_py_common/security_cipher.py @@ -0,0 +1,120 @@ +''' + +A common module for handling the encryption and +decryption of the feature passkey. It also takes +care of storying the secure cipher at root +protected file system + +''' + +import subprocess +import threading +import syslog +import os +from swsscommon.swsscommon import ConfigDBConnector + +class security_cipher: + _instance = None + _lock = threading.Lock() + + def __new__(cls): + with cls._lock: + if cls._instance is None: + cls._instance = super(security_cipher, cls).__new__(cls) + cls._instance._initialized = False + return cls._instance + + def __init__(self): + if not self._initialized: + self._file_path = "/etc/cipher_pass" + self._config_db = ConfigDBConnector() + self._config_db.connect() + # Note: Kept 1st index NA intentionally to map it with the cipher_pass file + # contents. The file has a comment at the 1st row / line + self._feature_list = ["NA", "TACPLUS", "RADIUS", "LDAP"] + if not os.path.exists(self._file_path): + with open(self._file_path, 'w') as file: + file.writelines("#Auto generated file for storing the encryption passwords\n") + file.writelines("TACPLUS : \nRADIUS : \nLDAP :\n") + #os.chmod(self._file_path, 0o644) + self._initialized = True + + # Write cipher_pass file + def __write_passwd_file(self, feature_type, passwd): + if feature_type == 'NA': + syslog.syslog(syslog.LOG_ERR, "__write_passwd_file: Invalid feature type: {}".format(feature_type)) + return + + if feature_type in self._feature_list: + try: + with open(self._file_path, 'r') as file: + lines = file.readlines() + # Update the password for given feature + lines[self._feature_list.index(feature_type)] = feature_type + ' : ' + passwd + '\n' + + #os.chmod(self._file_path, 0o777) + with open(self._file_path, 'w') as file: + file.writelines(lines) + #os.chmod(self._file_path, 0o644) + except FileNotFoundError: + syslog.syslog(syslog.LOG_ERR, "__write_passwd_file: File {} no found".format(self._file_path)) + except PermissionError: + syslog.syslog(syslog.LOG_ERR, "__write_passwd_file: Read permission denied: {}".format(self._file_path)) + + # Read cipher pass file and return the feature specifc + # password + def __read_passwd_file(self, feature_type): + passwd = None + if feature_type == 'NA': + syslog.syslog(syslog.LOG_ERR, "__read_passwd_file: Invalid feature type: {}".format(feature_type)) + return passwd + + if feature_type in self._feature_list: + try: + with open(self._file_path, "r") as file: + lines = file.readlines() + for line in lines: + if feature_type in line: + passwd = line.split(' : ')[1] + + except FileNotFoundError: + syslog.syslog(syslog.LOG_ERR, "__read_passwd_file: File {} no found".format(self._file_path)) + except PermissionError: + syslog.syslog(syslog.LOG_ERR, "__read_passwd_file: Read permission denied: {}".format(self._file_path)) + + syslog.syslog(syslog.LOG_ERR, "NIKHIL PASSWORD {}".format(passwd)) + return passwd + + # Encrypt the passkey + def encrypt_passkey(self, feature_type, secret, passwd): + cmd = [ 'openssl', 'enc', '-aes-128-cbc', '-A', '-a', '-salt', '-pbkdf2', '-pass', 'pass:' + passwd ] + p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True) + outsecret, errs = p.communicate(input=secret) + if not errs: + self.__write_passwd_file(feature_type, passwd) + + return outsecret,errs + + # Decrypt the passkey + def decrypt_passkey(self, feature_type, secret): + errs = "Passkey Decryption failed" + passwd = self.__read_passwd_file(feature_type) + if passwd is not None: + cmd = "echo " + format(secret) + " | openssl enc -aes-128-cbc -a -d -salt -pbkdf2 -pass pass:" + passwd + proc = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE) + output, errs = proc.communicate() + + if not errs: + output = output.decode('utf-8') + + return output, errs + + # Check if the encryption is enabled + def is_key_encrypt_enabled(self, table, entry): + key = 'key_encrypt' + data = self._config_db.get_entry(table, entry) + if data: + if key in data: + return data[key] + return False + From e6bb593fa6b233bb3285ba0f68fa3e2ac2c3eb2c Mon Sep 17 00:00:00 2001 From: nmoray Date: Thu, 16 Nov 2023 09:45:12 -0800 Subject: [PATCH 02/26] Added test cases to test security cipher module --- .../tests/test_security_cipher.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) create mode 100644 src/sonic-py-common/tests/test_security_cipher.py diff --git a/src/sonic-py-common/tests/test_security_cipher.py b/src/sonic-py-common/tests/test_security_cipher.py new file mode 100644 index 000000000000..0568300c4bcf --- /dev/null +++ b/src/sonic-py-common/tests/test_security_cipher.py @@ -0,0 +1,18 @@ +from sonic_py_common.secure_cipher import secure_cipher + +temp = security_cipher() + +class TestSecurityCipher(object): + def test_passkey_encryption(self): + encrypt, err = temp.encrypt_passkey("TACPLUS", "passkey1", "TEST1") + assert encrypt != "passkey1" + assert err == None + + def test_passkey_decryption(self): + encrypt, err = temp.encrypt_passkey("RADIUS", "passkey2", "TEST2") + assert err == None + decrypt, err = temp.decrypt_passkey("TACPLUS", encrypt) + assert err == None + assert decrypt == "passkey2" + + From f77490b596c07e63a6fb2468d4cada9f216de561 Mon Sep 17 00:00:00 2001 From: nmoray Date: Thu, 16 Nov 2023 09:59:38 -0800 Subject: [PATCH 03/26] Corrected typo --- src/sonic-py-common/tests/test_security_cipher.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/sonic-py-common/tests/test_security_cipher.py b/src/sonic-py-common/tests/test_security_cipher.py index 0568300c4bcf..26e271dc2365 100644 --- a/src/sonic-py-common/tests/test_security_cipher.py +++ b/src/sonic-py-common/tests/test_security_cipher.py @@ -11,7 +11,7 @@ def test_passkey_encryption(self): def test_passkey_decryption(self): encrypt, err = temp.encrypt_passkey("RADIUS", "passkey2", "TEST2") assert err == None - decrypt, err = temp.decrypt_passkey("TACPLUS", encrypt) + decrypt, err = temp.decrypt_passkey("RADIUS", encrypt) assert err == None assert decrypt == "passkey2" From 485aed8a75d4937634bbe6b400e68a9b7d639aae Mon Sep 17 00:00:00 2001 From: nmoray Date: Thu, 16 Nov 2023 22:26:27 -0800 Subject: [PATCH 04/26] Addressed comments and fixed AUT build errors --- src/sonic-py-common/sonic_py_common/security_cipher.py | 5 +---- src/sonic-py-common/tests/test_security_cipher.py | 2 +- 2 files changed, 2 insertions(+), 5 deletions(-) diff --git a/src/sonic-py-common/sonic_py_common/security_cipher.py b/src/sonic-py-common/sonic_py_common/security_cipher.py index 13b8b45b7546..7113ae631436 100644 --- a/src/sonic-py-common/sonic_py_common/security_cipher.py +++ b/src/sonic-py-common/sonic_py_common/security_cipher.py @@ -2,7 +2,7 @@ A common module for handling the encryption and decryption of the feature passkey. It also takes -care of storying the secure cipher at root +care of storing the secure cipher at root protected file system ''' @@ -36,7 +36,6 @@ def __init__(self): with open(self._file_path, 'w') as file: file.writelines("#Auto generated file for storing the encryption passwords\n") file.writelines("TACPLUS : \nRADIUS : \nLDAP :\n") - #os.chmod(self._file_path, 0o644) self._initialized = True # Write cipher_pass file @@ -52,10 +51,8 @@ def __write_passwd_file(self, feature_type, passwd): # Update the password for given feature lines[self._feature_list.index(feature_type)] = feature_type + ' : ' + passwd + '\n' - #os.chmod(self._file_path, 0o777) with open(self._file_path, 'w') as file: file.writelines(lines) - #os.chmod(self._file_path, 0o644) except FileNotFoundError: syslog.syslog(syslog.LOG_ERR, "__write_passwd_file: File {} no found".format(self._file_path)) except PermissionError: diff --git a/src/sonic-py-common/tests/test_security_cipher.py b/src/sonic-py-common/tests/test_security_cipher.py index 26e271dc2365..df74a2d08dda 100644 --- a/src/sonic-py-common/tests/test_security_cipher.py +++ b/src/sonic-py-common/tests/test_security_cipher.py @@ -1,4 +1,4 @@ -from sonic_py_common.secure_cipher import secure_cipher +from sonic_py_common.security_cipher import security_cipher temp = security_cipher() From 519dc8d88a6b36f9dda7ee0b81374a991c64e155 Mon Sep 17 00:00:00 2001 From: nmoray Date: Mon, 20 Nov 2023 03:21:20 -0800 Subject: [PATCH 05/26] Fixed build issues --- src/sonic-py-common/sonic_py_common/security_cipher.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/sonic-py-common/sonic_py_common/security_cipher.py b/src/sonic-py-common/sonic_py_common/security_cipher.py index 7113ae631436..9c8fa24467fc 100644 --- a/src/sonic-py-common/sonic_py_common/security_cipher.py +++ b/src/sonic-py-common/sonic_py_common/security_cipher.py @@ -16,6 +16,7 @@ class security_cipher: _instance = None _lock = threading.Lock() + _initialized = False def __new__(cls): with cls._lock: From cb51987430c310818564439eb6b2493ff225c792 Mon Sep 17 00:00:00 2001 From: nmoray Date: Wed, 22 Nov 2023 03:12:53 -0800 Subject: [PATCH 06/26] Fixed AUT issues --- src/sonic-py-common/tests/mock_swsscommon.py | 12 ++++++++++ .../tests/test_security_cipher.py | 22 +++++++++++-------- 2 files changed, 25 insertions(+), 9 deletions(-) diff --git a/src/sonic-py-common/tests/mock_swsscommon.py b/src/sonic-py-common/tests/mock_swsscommon.py index 8a619fe0e4c8..54e1e3bc0494 100644 --- a/src/sonic-py-common/tests/mock_swsscommon.py +++ b/src/sonic-py-common/tests/mock_swsscommon.py @@ -14,3 +14,15 @@ def connect(self, db): def get(self, db, table, field): return self.data.get(field, "N/A") + + +class ConfigDBConnector: + def __init__(self): + self.CONFIG_DB = 'CONFIG_DB' + self.data = {"key_encrypt": "True"} + + def connect(self, db): + pass + + def get(self, db, table, field): + return self.data.get(field, "N/A") diff --git a/src/sonic-py-common/tests/test_security_cipher.py b/src/sonic-py-common/tests/test_security_cipher.py index df74a2d08dda..0bc4a9dc2153 100644 --- a/src/sonic-py-common/tests/test_security_cipher.py +++ b/src/sonic-py-common/tests/test_security_cipher.py @@ -1,18 +1,22 @@ from sonic_py_common.security_cipher import security_cipher +from .mock_swsscommon import ConfigDBConnector -temp = security_cipher() class TestSecurityCipher(object): def test_passkey_encryption(self): - encrypt, err = temp.encrypt_passkey("TACPLUS", "passkey1", "TEST1") - assert encrypt != "passkey1" - assert err == None + with mock.patch("sonic_py_common.security_cipher.ConfigDBConnector", new=ConfigDBConnector): + temp = security_cipher() + encrypt, err = temp.encrypt_passkey("TACPLUS", "passkey1", "TEST1") + assert encrypt != "passkey1" + assert err == None def test_passkey_decryption(self): - encrypt, err = temp.encrypt_passkey("RADIUS", "passkey2", "TEST2") - assert err == None - decrypt, err = temp.decrypt_passkey("RADIUS", encrypt) - assert err == None - assert decrypt == "passkey2" + with mock.patch("sonic_py_common.security_cipher.ConfigDBConnector", new=ConfigDBConnector): + temp = security_cipher() + encrypt, err = temp.encrypt_passkey("RADIUS", "passkey2", "TEST2") + assert err == None + decrypt, err = temp.decrypt_passkey("RADIUS", encrypt) + assert err == None + assert decrypt == "passkey2" From f4745b451846748129fafe55bd3b3662d6f7b1a4 Mon Sep 17 00:00:00 2001 From: nmoray Date: Wed, 22 Nov 2023 03:55:48 -0800 Subject: [PATCH 07/26] Fixed indentation issue --- src/sonic-py-common/tests/mock_swsscommon.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/sonic-py-common/tests/mock_swsscommon.py b/src/sonic-py-common/tests/mock_swsscommon.py index 54e1e3bc0494..e6c62272297b 100644 --- a/src/sonic-py-common/tests/mock_swsscommon.py +++ b/src/sonic-py-common/tests/mock_swsscommon.py @@ -17,7 +17,7 @@ def get(self, db, table, field): class ConfigDBConnector: - def __init__(self): + def __init__(self): self.CONFIG_DB = 'CONFIG_DB' self.data = {"key_encrypt": "True"} From 2b08431cc13dcab56506843606e2656512e861d4 Mon Sep 17 00:00:00 2001 From: nmoray Date: Wed, 22 Nov 2023 05:04:54 -0800 Subject: [PATCH 08/26] Fixed build issues --- src/sonic-py-common/tests/test_security_cipher.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/src/sonic-py-common/tests/test_security_cipher.py b/src/sonic-py-common/tests/test_security_cipher.py index 0bc4a9dc2153..5e69cb63621f 100644 --- a/src/sonic-py-common/tests/test_security_cipher.py +++ b/src/sonic-py-common/tests/test_security_cipher.py @@ -1,5 +1,12 @@ +import sys + +if sys.version_info.major == 3: + from unittest import mock +else: + import mock + from sonic_py_common.security_cipher import security_cipher -from .mock_swsscommon import ConfigDBConnector +from .mock_swsscommon import ConfigDBConnector class TestSecurityCipher(object): From 8129f8669107750726486c9068be224bf04aac87 Mon Sep 17 00:00:00 2001 From: nmoray Date: Wed, 22 Nov 2023 06:06:50 -0800 Subject: [PATCH 09/26] Fixed build issues --- src/sonic-py-common/tests/mock_swsscommon.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/sonic-py-common/tests/mock_swsscommon.py b/src/sonic-py-common/tests/mock_swsscommon.py index e6c62272297b..567a5615503b 100644 --- a/src/sonic-py-common/tests/mock_swsscommon.py +++ b/src/sonic-py-common/tests/mock_swsscommon.py @@ -21,7 +21,7 @@ def __init__(self): self.CONFIG_DB = 'CONFIG_DB' self.data = {"key_encrypt": "True"} - def connect(self, db): + def connect(self): pass def get(self, db, table, field): From 8dfacd3c80e54e61f2365f4f17c29a0fcdf4d8c1 Mon Sep 17 00:00:00 2001 From: nmoray Date: Wed, 22 Nov 2023 09:21:07 -0800 Subject: [PATCH 10/26] Added support of mock class in existing AUT --- .../tests/test_security_cipher.py | 27 +++++++++++++------ 1 file changed, 19 insertions(+), 8 deletions(-) diff --git a/src/sonic-py-common/tests/test_security_cipher.py b/src/sonic-py-common/tests/test_security_cipher.py index 5e69cb63621f..61ba15b57c8e 100644 --- a/src/sonic-py-common/tests/test_security_cipher.py +++ b/src/sonic-py-common/tests/test_security_cipher.py @@ -5,25 +5,36 @@ else: import mock +import pytest from sonic_py_common.security_cipher import security_cipher from .mock_swsscommon import ConfigDBConnector +EXPECTED_PASSWD = "TEST2" class TestSecurityCipher(object): def test_passkey_encryption(self): with mock.patch("sonic_py_common.security_cipher.ConfigDBConnector", new=ConfigDBConnector): temp = security_cipher() - encrypt, err = temp.encrypt_passkey("TACPLUS", "passkey1", "TEST1") - assert encrypt != "passkey1" - assert err == None + + # Use patch to replace the built-in 'open' function with a mock + with mock.patch("builtins.open", mock_open()) as mock_file: + encrypt, err = temp.encrypt_passkey("TACPLUS", "passkey1", "TEST1") + assert encrypt != "passkey1" + assert err == None def test_passkey_decryption(self): with mock.patch("sonic_py_common.security_cipher.ConfigDBConnector", new=ConfigDBConnector): temp = security_cipher() - encrypt, err = temp.encrypt_passkey("RADIUS", "passkey2", "TEST2") - assert err == None - decrypt, err = temp.decrypt_passkey("RADIUS", encrypt) - assert err == None - assert decrypt == "passkey2" + + # Use patch to replace the built-in 'open' function with a mock + with mock.patch("builtins.open", mock_open()) as mock_file: + encrypt, err = temp.encrypt_passkey("RADIUS", "passkey2", EXPECTED_PASSWD) + assert err == None + + # Use patch to replace the built-in 'open' function with a mock + with mock.patch("builtins.open", mock_open(read_data=EXPECTED_PASSWD)) as mock_file: + decrypt, err = temp.decrypt_passkey("RADIUS", encrypt) + assert err == None + assert decrypt == "passkey2" From c6fd0c9c77a8e9d0d6d96735de9b104b4ed29018 Mon Sep 17 00:00:00 2001 From: nmoray Date: Wed, 22 Nov 2023 20:51:34 -0800 Subject: [PATCH 11/26] fixed build issues --- src/sonic-py-common/tests/test_security_cipher.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/sonic-py-common/tests/test_security_cipher.py b/src/sonic-py-common/tests/test_security_cipher.py index 61ba15b57c8e..87670cb722a7 100644 --- a/src/sonic-py-common/tests/test_security_cipher.py +++ b/src/sonic-py-common/tests/test_security_cipher.py @@ -13,7 +13,8 @@ class TestSecurityCipher(object): def test_passkey_encryption(self): - with mock.patch("sonic_py_common.security_cipher.ConfigDBConnector", new=ConfigDBConnector): + with mock.patch("sonic_py_common.security_cipher.ConfigDBConnector", new=ConfigDBConnector), \ + mock.patch("builtins.open", mock_open()) as mock_file: temp = security_cipher() # Use patch to replace the built-in 'open' function with a mock @@ -23,7 +24,8 @@ def test_passkey_encryption(self): assert err == None def test_passkey_decryption(self): - with mock.patch("sonic_py_common.security_cipher.ConfigDBConnector", new=ConfigDBConnector): + with mock.patch("sonic_py_common.security_cipher.ConfigDBConnector", new=ConfigDBConnector), \ + mock.patch("builtins.open", mock_open()) as mock_file: temp = security_cipher() # Use patch to replace the built-in 'open' function with a mock From 547b31fd50a4fcf3f9b427171d39b0f6f8b237a4 Mon Sep 17 00:00:00 2001 From: nmoray Date: Wed, 22 Nov 2023 21:41:10 -0800 Subject: [PATCH 12/26] Fixed build issues --- src/sonic-py-common/tests/test_security_cipher.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/sonic-py-common/tests/test_security_cipher.py b/src/sonic-py-common/tests/test_security_cipher.py index 87670cb722a7..9c3f5382e456 100644 --- a/src/sonic-py-common/tests/test_security_cipher.py +++ b/src/sonic-py-common/tests/test_security_cipher.py @@ -14,27 +14,27 @@ class TestSecurityCipher(object): def test_passkey_encryption(self): with mock.patch("sonic_py_common.security_cipher.ConfigDBConnector", new=ConfigDBConnector), \ - mock.patch("builtins.open", mock_open()) as mock_file: + mock.patch("builtins.open",mock.mock_open()) as mock_file: temp = security_cipher() # Use patch to replace the built-in 'open' function with a mock - with mock.patch("builtins.open", mock_open()) as mock_file: + with mock.patch("builtins.open", mock.mock_open()) as mock_file: encrypt, err = temp.encrypt_passkey("TACPLUS", "passkey1", "TEST1") assert encrypt != "passkey1" assert err == None def test_passkey_decryption(self): with mock.patch("sonic_py_common.security_cipher.ConfigDBConnector", new=ConfigDBConnector), \ - mock.patch("builtins.open", mock_open()) as mock_file: + mock.patch("builtins.open", mock.mock_open()) as mock_file: temp = security_cipher() # Use patch to replace the built-in 'open' function with a mock - with mock.patch("builtins.open", mock_open()) as mock_file: + with mock.patch("builtins.open", mock.mock_open()) as mock_file: encrypt, err = temp.encrypt_passkey("RADIUS", "passkey2", EXPECTED_PASSWD) assert err == None # Use patch to replace the built-in 'open' function with a mock - with mock.patch("builtins.open", mock_open(read_data=EXPECTED_PASSWD)) as mock_file: + with mock.patch("builtins.open", mock.mock_open(read_data=EXPECTED_PASSWD)) as mock_file: decrypt, err = temp.decrypt_passkey("RADIUS", encrypt) assert err == None assert decrypt == "passkey2" From 7a8c120e8f686ca1317fa782e755b61a3d0865f8 Mon Sep 17 00:00:00 2001 From: nmoray Date: Wed, 22 Nov 2023 23:32:46 -0800 Subject: [PATCH 13/26] fixed build issues --- src/sonic-py-common/tests/test_security_cipher.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/sonic-py-common/tests/test_security_cipher.py b/src/sonic-py-common/tests/test_security_cipher.py index 9c3f5382e456..4e6aece4e322 100644 --- a/src/sonic-py-common/tests/test_security_cipher.py +++ b/src/sonic-py-common/tests/test_security_cipher.py @@ -9,6 +9,12 @@ from sonic_py_common.security_cipher import security_cipher from .mock_swsscommon import ConfigDBConnector +# TODO: Remove this if/else block once we no longer support Python 2 +if sys.version_info.major == 3: + BUILTINS = "builtins" +else: + BUILTINS = "__builtin__" + EXPECTED_PASSWD = "TEST2" class TestSecurityCipher(object): From 78d66c7a0c25fbf0b58d8d7b451f46e26edab2e2 Mon Sep 17 00:00:00 2001 From: nmoray Date: Thu, 23 Nov 2023 00:34:22 -0800 Subject: [PATCH 14/26] Fixed build issues --- src/sonic-py-common/tests/test_security_cipher.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/sonic-py-common/tests/test_security_cipher.py b/src/sonic-py-common/tests/test_security_cipher.py index 4e6aece4e322..b91dcfbdca26 100644 --- a/src/sonic-py-common/tests/test_security_cipher.py +++ b/src/sonic-py-common/tests/test_security_cipher.py @@ -20,27 +20,27 @@ class TestSecurityCipher(object): def test_passkey_encryption(self): with mock.patch("sonic_py_common.security_cipher.ConfigDBConnector", new=ConfigDBConnector), \ - mock.patch("builtins.open",mock.mock_open()) as mock_file: + mock.patch("{}.open".format(BUILTINS),mock.mock_open()) as mock_file: temp = security_cipher() # Use patch to replace the built-in 'open' function with a mock - with mock.patch("builtins.open", mock.mock_open()) as mock_file: + with mock.patch("{}.open".format(BUILTINS), mock.mock_open()) as mock_file: encrypt, err = temp.encrypt_passkey("TACPLUS", "passkey1", "TEST1") assert encrypt != "passkey1" assert err == None def test_passkey_decryption(self): with mock.patch("sonic_py_common.security_cipher.ConfigDBConnector", new=ConfigDBConnector), \ - mock.patch("builtins.open", mock.mock_open()) as mock_file: + mock.patch("{}.open".format(BUILTINS), mock.mock_open()) as mock_file: temp = security_cipher() # Use patch to replace the built-in 'open' function with a mock - with mock.patch("builtins.open", mock.mock_open()) as mock_file: + with mock.patch("{}.open".format(BUILTINS), mock.mock_open()) as mock_file: encrypt, err = temp.encrypt_passkey("RADIUS", "passkey2", EXPECTED_PASSWD) assert err == None # Use patch to replace the built-in 'open' function with a mock - with mock.patch("builtins.open", mock.mock_open(read_data=EXPECTED_PASSWD)) as mock_file: + with mock.patch("{}.open".format(BUILTINS), mock.mock_open(read_data=EXPECTED_PASSWD)) as mock_file: decrypt, err = temp.decrypt_passkey("RADIUS", encrypt) assert err == None assert decrypt == "passkey2" From 6ee2c7e73c6691470bfb3aca5d26eac12a54530d Mon Sep 17 00:00:00 2001 From: nmoray Date: Thu, 23 Nov 2023 08:05:31 -0800 Subject: [PATCH 15/26] fixed build issues --- .../tests/test_security_cipher.py | 29 +++++++++++++++++-- 1 file changed, 26 insertions(+), 3 deletions(-) diff --git a/src/sonic-py-common/tests/test_security_cipher.py b/src/sonic-py-common/tests/test_security_cipher.py index b91dcfbdca26..022d25a3df61 100644 --- a/src/sonic-py-common/tests/test_security_cipher.py +++ b/src/sonic-py-common/tests/test_security_cipher.py @@ -15,7 +15,20 @@ else: BUILTINS = "__builtin__" -EXPECTED_PASSWD = "TEST2" +DEFAULT_FILE = [ + "#Auto generated file for storing the encryption passwords", + "TACPLUS : ", + "RADIUS : ", + "LDAP :" + ] + +UPDATED_FILE = [ + "#Auto generated file for storing the encryption passwords", + "TACPLUS : ", + "RADIUS : TEST2", + "LDAP :" + ] + class TestSecurityCipher(object): def test_passkey_encryption(self): @@ -25,6 +38,9 @@ def test_passkey_encryption(self): # Use patch to replace the built-in 'open' function with a mock with mock.patch("{}.open".format(BUILTINS), mock.mock_open()) as mock_file: + mock_fd = mock.MagicMock() + mock_fd.readlines = mock.MagicMock(return_value=DEFAULT_FILE) + mock_file.return_value.__enter__.return_value = mock_fd encrypt, err = temp.encrypt_passkey("TACPLUS", "passkey1", "TEST1") assert encrypt != "passkey1" assert err == None @@ -36,11 +52,18 @@ def test_passkey_decryption(self): # Use patch to replace the built-in 'open' function with a mock with mock.patch("{}.open".format(BUILTINS), mock.mock_open()) as mock_file: - encrypt, err = temp.encrypt_passkey("RADIUS", "passkey2", EXPECTED_PASSWD) + mock_fd = mock.MagicMock() + mock_fd.readlines = mock.MagicMock(return_value=DEFAULT_FILE) + mock_file.return_value.__enter__.return_value = mock_fd + encrypt, err = temp.encrypt_passkey("RADIUS", "passkey2", "TEST2") assert err == None # Use patch to replace the built-in 'open' function with a mock - with mock.patch("{}.open".format(BUILTINS), mock.mock_open(read_data=EXPECTED_PASSWD)) as mock_file: + #with mock.patch("{}.open".format(BUILTINS), mock.mock_open(read_data=EXPECTED_PASSWD)) as mock_file: + with mock.patch("{}.open".format(BUILTINS), mock.mock_open()) as mock_file: + mock_fd = mock.MagicMock() + mock_fd.readlines = mock.MagicMock(return_value=UPDATED_FILE) + mock_file.return_value.__enter__.return_value = mock_fd decrypt, err = temp.decrypt_passkey("RADIUS", encrypt) assert err == None assert decrypt == "passkey2" From 0fd2ba089031a8875dbb903b94c91bc0d6ac03e8 Mon Sep 17 00:00:00 2001 From: nmoray Date: Thu, 23 Nov 2023 09:06:13 -0800 Subject: [PATCH 16/26] fixed build issues --- src/sonic-py-common/tests/test_security_cipher.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/src/sonic-py-common/tests/test_security_cipher.py b/src/sonic-py-common/tests/test_security_cipher.py index 022d25a3df61..af6b08d2b9f0 100644 --- a/src/sonic-py-common/tests/test_security_cipher.py +++ b/src/sonic-py-common/tests/test_security_cipher.py @@ -43,7 +43,6 @@ def test_passkey_encryption(self): mock_file.return_value.__enter__.return_value = mock_fd encrypt, err = temp.encrypt_passkey("TACPLUS", "passkey1", "TEST1") assert encrypt != "passkey1" - assert err == None def test_passkey_decryption(self): with mock.patch("sonic_py_common.security_cipher.ConfigDBConnector", new=ConfigDBConnector), \ @@ -56,7 +55,6 @@ def test_passkey_decryption(self): mock_fd.readlines = mock.MagicMock(return_value=DEFAULT_FILE) mock_file.return_value.__enter__.return_value = mock_fd encrypt, err = temp.encrypt_passkey("RADIUS", "passkey2", "TEST2") - assert err == None # Use patch to replace the built-in 'open' function with a mock #with mock.patch("{}.open".format(BUILTINS), mock.mock_open(read_data=EXPECTED_PASSWD)) as mock_file: @@ -65,7 +63,6 @@ def test_passkey_decryption(self): mock_fd.readlines = mock.MagicMock(return_value=UPDATED_FILE) mock_file.return_value.__enter__.return_value = mock_fd decrypt, err = temp.decrypt_passkey("RADIUS", encrypt) - assert err == None assert decrypt == "passkey2" From aff2df14bd6d5cab1907036a1471cbe6e36ce625 Mon Sep 17 00:00:00 2001 From: nmoray Date: Fri, 24 Nov 2023 03:36:13 -0800 Subject: [PATCH 17/26] Updated passkey leaf length to work with encrypted string too --- src/sonic-yang-models/yang-models/sonic-system-tacacs.yang | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/sonic-yang-models/yang-models/sonic-system-tacacs.yang b/src/sonic-yang-models/yang-models/sonic-system-tacacs.yang index 6abbcf3523b0..007b236f22a2 100644 --- a/src/sonic-yang-models/yang-models/sonic-system-tacacs.yang +++ b/src/sonic-yang-models/yang-models/sonic-system-tacacs.yang @@ -93,7 +93,7 @@ module sonic-system-tacacs { leaf passkey { type string { - length "1..65"; + length "1..256"; pattern "[^ #,]*" { error-message 'TACACS shared secret (Valid chars are ASCII printable except SPACE, "#", and ",")'; } From 55c7cc736092390db746af7863d96be11b6c5bec Mon Sep 17 00:00:00 2001 From: nmoray Date: Wed, 31 Jan 2024 22:00:05 -0800 Subject: [PATCH 18/26] Added a delete function for removing the footprint of cipher_pass file in case encryption is disabled --- .../sonic_py_common/security_cipher.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/src/sonic-py-common/sonic_py_common/security_cipher.py b/src/sonic-py-common/sonic_py_common/security_cipher.py index 9c8fa24467fc..2a4c0c88df5e 100644 --- a/src/sonic-py-common/sonic_py_common/security_cipher.py +++ b/src/sonic-py-common/sonic_py_common/security_cipher.py @@ -116,3 +116,15 @@ def is_key_encrypt_enabled(self, table, entry): return data[key] return False + def del_cipher_pass(self): + try: + # Check if the file exists + if os.path.exists(self._file_path): + # Attempt to delete the file + os.remove(self._file_path) + syslog.syslog(syslog.LOG_INFO, "del_cipher_pass: {} file has been removed".format((self._file_path))) + else: + syslog.syslog(syslog.LOG_INFO, "del_cipher_pass: {} file doesn't exist".format((self._file_path))) + except Exception as e: + syslog.syslog(syslog.LOG_ERR, "del_cipher_pass: {} Exception occurred: {}".format((e))) + From 82680212bedc77f6f6f8fc6c311676109a675e0b Mon Sep 17 00:00:00 2001 From: nmoray Date: Thu, 1 Feb 2024 01:25:38 -0800 Subject: [PATCH 19/26] Change the access permission of cipher_pass file from 644 to 640 (-rw-r-----) --- src/sonic-py-common/sonic_py_common/security_cipher.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/sonic-py-common/sonic_py_common/security_cipher.py b/src/sonic-py-common/sonic_py_common/security_cipher.py index 2a4c0c88df5e..62130a3fcfbd 100644 --- a/src/sonic-py-common/sonic_py_common/security_cipher.py +++ b/src/sonic-py-common/sonic_py_common/security_cipher.py @@ -37,6 +37,7 @@ def __init__(self): with open(self._file_path, 'w') as file: file.writelines("#Auto generated file for storing the encryption passwords\n") file.writelines("TACPLUS : \nRADIUS : \nLDAP :\n") + os.chmod(self._file_path, 0o640) self._initialized = True # Write cipher_pass file @@ -52,8 +53,10 @@ def __write_passwd_file(self, feature_type, passwd): # Update the password for given feature lines[self._feature_list.index(feature_type)] = feature_type + ' : ' + passwd + '\n' + os.chmod(self._file_path, 0o777) with open(self._file_path, 'w') as file: file.writelines(lines) + os.chmod(self._file_path, 0o640) except FileNotFoundError: syslog.syslog(syslog.LOG_ERR, "__write_passwd_file: File {} no found".format(self._file_path)) except PermissionError: @@ -69,12 +72,13 @@ def __read_passwd_file(self, feature_type): if feature_type in self._feature_list: try: + os.chmod(self._file_path, 0o644) with open(self._file_path, "r") as file: lines = file.readlines() for line in lines: if feature_type in line: passwd = line.split(' : ')[1] - + os.chmod(self._file_path, 0o640) except FileNotFoundError: syslog.syslog(syslog.LOG_ERR, "__read_passwd_file: File {} no found".format(self._file_path)) except PermissionError: From 2fbec21c40718597c07c71d28ea41b644941c147 Mon Sep 17 00:00:00 2001 From: nmoray Date: Thu, 1 Feb 2024 08:20:51 -0800 Subject: [PATCH 20/26] updated scripts to mock chmod --- src/sonic-py-common/tests/test_security_cipher.py | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/src/sonic-py-common/tests/test_security_cipher.py b/src/sonic-py-common/tests/test_security_cipher.py index af6b08d2b9f0..1bda77f73d3d 100644 --- a/src/sonic-py-common/tests/test_security_cipher.py +++ b/src/sonic-py-common/tests/test_security_cipher.py @@ -33,11 +33,13 @@ class TestSecurityCipher(object): def test_passkey_encryption(self): with mock.patch("sonic_py_common.security_cipher.ConfigDBConnector", new=ConfigDBConnector), \ - mock.patch("{}.open".format(BUILTINS),mock.mock_open()) as mock_file: + mock.patch("os.chmod") as mock_chmod, \ + mock.patch("{}.open".format(BUILTINS),mock.mock_open()) as mock_file: temp = security_cipher() # Use patch to replace the built-in 'open' function with a mock - with mock.patch("{}.open".format(BUILTINS), mock.mock_open()) as mock_file: + with mock.patch("{}.open".format(BUILTINS), mock.mock_open()) as mock_file, \ + mock.patch("os.chmod") as mock_chmod: mock_fd = mock.MagicMock() mock_fd.readlines = mock.MagicMock(return_value=DEFAULT_FILE) mock_file.return_value.__enter__.return_value = mock_fd @@ -46,11 +48,13 @@ def test_passkey_encryption(self): def test_passkey_decryption(self): with mock.patch("sonic_py_common.security_cipher.ConfigDBConnector", new=ConfigDBConnector), \ - mock.patch("{}.open".format(BUILTINS), mock.mock_open()) as mock_file: + mock.patch("os.chmod") as mock_chmod, \ + mock.patch("{}.open".format(BUILTINS), mock.mock_open()) as mock_file: temp = security_cipher() # Use patch to replace the built-in 'open' function with a mock - with mock.patch("{}.open".format(BUILTINS), mock.mock_open()) as mock_file: + with mock.patch("{}.open".format(BUILTINS), mock.mock_open()) as mock_file, \ + mock.patch("os.chmod") as mock_chmod: mock_fd = mock.MagicMock() mock_fd.readlines = mock.MagicMock(return_value=DEFAULT_FILE) mock_file.return_value.__enter__.return_value = mock_fd @@ -58,7 +62,8 @@ def test_passkey_decryption(self): # Use patch to replace the built-in 'open' function with a mock #with mock.patch("{}.open".format(BUILTINS), mock.mock_open(read_data=EXPECTED_PASSWD)) as mock_file: - with mock.patch("{}.open".format(BUILTINS), mock.mock_open()) as mock_file: + with mock.patch("{}.open".format(BUILTINS), mock.mock_open()) as mock_file, \ + mock.patch("os.chmod") as mock_chmod: mock_fd = mock.MagicMock() mock_fd.readlines = mock.MagicMock(return_value=UPDATED_FILE) mock_file.return_value.__enter__.return_value = mock_fd From 95a4c10020909d76549496ec4ffa918b28bd6c29 Mon Sep 17 00:00:00 2001 From: nmoray Date: Thu, 1 Feb 2024 20:08:55 -0800 Subject: [PATCH 21/26] Removed unwated debugs --- src/sonic-py-common/sonic_py_common/security_cipher.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/sonic-py-common/sonic_py_common/security_cipher.py b/src/sonic-py-common/sonic_py_common/security_cipher.py index 62130a3fcfbd..d40c72c7e884 100644 --- a/src/sonic-py-common/sonic_py_common/security_cipher.py +++ b/src/sonic-py-common/sonic_py_common/security_cipher.py @@ -84,7 +84,6 @@ def __read_passwd_file(self, feature_type): except PermissionError: syslog.syslog(syslog.LOG_ERR, "__read_passwd_file: Read permission denied: {}".format(self._file_path)) - syslog.syslog(syslog.LOG_ERR, "NIKHIL PASSWORD {}".format(passwd)) return passwd # Encrypt the passkey From 719053b2ca392fe1c25953aea2eb695705a7510c Mon Sep 17 00:00:00 2001 From: nmoray Date: Thu, 11 Apr 2024 14:15:07 +0000 Subject: [PATCH 22/26] renamed class name to master_key_mgr --- src/sonic-py-common/sonic_py_common/security_cipher.py | 4 ++-- src/sonic-py-common/tests/test_security_cipher.py | 6 +++--- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/src/sonic-py-common/sonic_py_common/security_cipher.py b/src/sonic-py-common/sonic_py_common/security_cipher.py index d40c72c7e884..711186521f26 100644 --- a/src/sonic-py-common/sonic_py_common/security_cipher.py +++ b/src/sonic-py-common/sonic_py_common/security_cipher.py @@ -13,7 +13,7 @@ import os from swsscommon.swsscommon import ConfigDBConnector -class security_cipher: +class master_key_mgr: _instance = None _lock = threading.Lock() _initialized = False @@ -21,7 +21,7 @@ class security_cipher: def __new__(cls): with cls._lock: if cls._instance is None: - cls._instance = super(security_cipher, cls).__new__(cls) + cls._instance = super(master_key_mgr, cls).__new__(cls) cls._instance._initialized = False return cls._instance diff --git a/src/sonic-py-common/tests/test_security_cipher.py b/src/sonic-py-common/tests/test_security_cipher.py index 1bda77f73d3d..c3f169a873e3 100644 --- a/src/sonic-py-common/tests/test_security_cipher.py +++ b/src/sonic-py-common/tests/test_security_cipher.py @@ -6,7 +6,7 @@ import mock import pytest -from sonic_py_common.security_cipher import security_cipher +from sonic_py_common.security_cipher import master_key_mgr from .mock_swsscommon import ConfigDBConnector # TODO: Remove this if/else block once we no longer support Python 2 @@ -35,7 +35,7 @@ def test_passkey_encryption(self): with mock.patch("sonic_py_common.security_cipher.ConfigDBConnector", new=ConfigDBConnector), \ mock.patch("os.chmod") as mock_chmod, \ mock.patch("{}.open".format(BUILTINS),mock.mock_open()) as mock_file: - temp = security_cipher() + temp = master_key_mgr() # Use patch to replace the built-in 'open' function with a mock with mock.patch("{}.open".format(BUILTINS), mock.mock_open()) as mock_file, \ @@ -50,7 +50,7 @@ def test_passkey_decryption(self): with mock.patch("sonic_py_common.security_cipher.ConfigDBConnector", new=ConfigDBConnector), \ mock.patch("os.chmod") as mock_chmod, \ mock.patch("{}.open".format(BUILTINS), mock.mock_open()) as mock_file: - temp = security_cipher() + temp = master_key_mgr() # Use patch to replace the built-in 'open' function with a mock with mock.patch("{}.open".format(BUILTINS), mock.mock_open()) as mock_file, \ From 1ab70671b15e841464faccce3c2a00446ef32f7a Mon Sep 17 00:00:00 2001 From: nmoray Date: Mon, 10 Mar 2025 13:29:06 +0000 Subject: [PATCH 23/26] Addressed new comments --- .../sonic_py_common/security_cipher.py | 34 +++++++++++++------ .../yang-models/sonic-system-tacacs.yang | 7 +++- 2 files changed, 30 insertions(+), 11 deletions(-) diff --git a/src/sonic-py-common/sonic_py_common/security_cipher.py b/src/sonic-py-common/sonic_py_common/security_cipher.py index 711186521f26..ba71bb6336f7 100644 --- a/src/sonic-py-common/sonic_py_common/security_cipher.py +++ b/src/sonic-py-common/sonic_py_common/security_cipher.py @@ -36,7 +36,8 @@ def __init__(self): if not os.path.exists(self._file_path): with open(self._file_path, 'w') as file: file.writelines("#Auto generated file for storing the encryption passwords\n") - file.writelines("TACPLUS : \nRADIUS : \nLDAP :\n") + for feature in self._feature_list[1:]: # Skip the first "NA" entry + file.write(f"{feature} : \n") os.chmod(self._file_path, 0o640) self._initialized = True @@ -119,15 +120,28 @@ def is_key_encrypt_enabled(self, table, entry): return data[key] return False - def del_cipher_pass(self): + + def del_cipher_pass(self, feature_type): + """ + Removes only the password for the given feature_type while keeping the file structure intact. + """ try: - # Check if the file exists - if os.path.exists(self._file_path): - # Attempt to delete the file - os.remove(self._file_path) - syslog.syslog(syslog.LOG_INFO, "del_cipher_pass: {} file has been removed".format((self._file_path))) - else: - syslog.syslog(syslog.LOG_INFO, "del_cipher_pass: {} file doesn't exist".format((self._file_path))) + os.chmod(self._file_path, 0o777) + with open(self._file_path, "r") as file: + lines = file.readlines() + + updated_lines = [] + for line in lines: + if line.strip().startswith(f"{feature_type} :"): + updated_lines.append(f"{feature_type} : \n") # Remove password but keep format + else: + updated_lines.append(line) + + with open(self._file_path, 'w') as file: + file.writelines(updated_lines) + os.chmod(self._file_path, 0o640) + + syslog.syslog(syslog.LOG_INFO, "del_cipher_pass: Password for {} has been removed".format((feature_type))) + except Exception as e: syslog.syslog(syslog.LOG_ERR, "del_cipher_pass: {} Exception occurred: {}".format((e))) - diff --git a/src/sonic-yang-models/yang-models/sonic-system-tacacs.yang b/src/sonic-yang-models/yang-models/sonic-system-tacacs.yang index 007b236f22a2..e8a113954ee2 100644 --- a/src/sonic-yang-models/yang-models/sonic-system-tacacs.yang +++ b/src/sonic-yang-models/yang-models/sonic-system-tacacs.yang @@ -131,9 +131,14 @@ module sonic-system-tacacs { default 5; } + leaf key_encrypt { + type boolean; + description "Indicates if the passkey is encrypted."; + } + leaf passkey { type string { - length "1..65"; + length "1..256"; pattern "[^ #,]*" { error-message 'TACACS shared secret (Valid chars are ASCII printable except SPACE, "#", and ",")'; } From 5a94a62e7369bbc9272e97d41c464e1cfd1eb122 Mon Sep 17 00:00:00 2001 From: nmoray Date: Mon, 19 May 2025 12:57:44 +0000 Subject: [PATCH 24/26] Fixed semgrep findings --- .../sonic_py_common/security_cipher.py | 77 ++++++++++++++----- 1 file changed, 57 insertions(+), 20 deletions(-) diff --git a/src/sonic-py-common/sonic_py_common/security_cipher.py b/src/sonic-py-common/sonic_py_common/security_cipher.py index ba71bb6336f7..d48d29cf7fd6 100644 --- a/src/sonic-py-common/sonic_py_common/security_cipher.py +++ b/src/sonic-py-common/sonic_py_common/security_cipher.py @@ -11,6 +11,7 @@ import threading import syslog import os +import base64 from swsscommon.swsscommon import ConfigDBConnector class master_key_mgr: @@ -54,7 +55,7 @@ def __write_passwd_file(self, feature_type, passwd): # Update the password for given feature lines[self._feature_list.index(feature_type)] = feature_type + ' : ' + passwd + '\n' - os.chmod(self._file_path, 0o777) + os.chmod(self._file_path, 0o640) with open(self._file_path, 'w') as file: file.writelines(lines) os.chmod(self._file_path, 0o640) @@ -63,6 +64,7 @@ def __write_passwd_file(self, feature_type, passwd): except PermissionError: syslog.syslog(syslog.LOG_ERR, "__write_passwd_file: Read permission denied: {}".format(self._file_path)) + # Read cipher pass file and return the feature specifc # password def __read_passwd_file(self, feature_type): @@ -87,29 +89,63 @@ def __read_passwd_file(self, feature_type): return passwd - # Encrypt the passkey - def encrypt_passkey(self, feature_type, secret, passwd): - cmd = [ 'openssl', 'enc', '-aes-128-cbc', '-A', '-a', '-salt', '-pbkdf2', '-pass', 'pass:' + passwd ] - p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True) - outsecret, errs = p.communicate(input=secret) - if not errs: + + def encrypt_passkey(self, feature_type, secret: str, passwd: str) -> str: + """ + Encrypts the plaintext using OpenSSL (AES-128-CBC, with salt and pbkdf2, no base64) + and returns the result as a hex string. + """ + cmd = [ + "openssl", "enc", "-aes-128-cbc", "-salt", "-pbkdf2", + "-pass", f"pass:{passwd}" + ] + try: + result = subprocess.run( + cmd, + input=secret.encode(), + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=True + ) + encrypted_bytes = result.stdout + b64_encoded = base64.b64encode(encrypted_bytes).decode() self.__write_passwd_file(feature_type, passwd) + return b64_encoded + except subprocess.CalledProcessError as e: + syslog.syslog(syslog.LOG_ERR, "encrypt_passkey: {} Encryption failed with ERR: {}".format((e))) + return "" - return outsecret,errs - # Decrypt the passkey - def decrypt_passkey(self, feature_type, secret): - errs = "Passkey Decryption failed" - passwd = self.__read_passwd_file(feature_type) - if passwd is not None: - cmd = "echo " + format(secret) + " | openssl enc -aes-128-cbc -a -d -salt -pbkdf2 -pass pass:" + passwd - proc = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE) - output, errs = proc.communicate() + def decrypt_passkey(self, feature_type, b64_encoded: str) -> str: + """ + Decrypts a hex-encoded encrypted string using OpenSSL (AES-128-CBC, with salt and pbkdf2, no base64). + Returns the decrypted plaintext. + """ + + passwd = self.__read_passwd_file(feature_type).strip() + if passwd is None: + syslog.syslog(syslog.LOG_ERR, "decrypt_passkey: Enpty password for {} feature type".format(feature_type)) + return "" - if not errs: - output = output.decode('utf-8') + try: + encrypted_bytes = base64.b64decode(b64_encoded) + + cmd = [ + "openssl", "enc", "-aes-128-cbc", "-d", "-salt", "-pbkdf2", + "-pass", f"pass:{passwd}" + ] + result = subprocess.run( + cmd, + input=encrypted_bytes, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=True + ) + return result.stdout.decode().strip() + except subprocess.CalledProcessError as e: + syslog.syslog(syslog.LOG_ERR, "decrypt_passkey: Decryption failed with an ERR: {}".format(e.stderr.decode())) + return "" - return output, errs # Check if the encryption is enabled def is_key_encrypt_enabled(self, table, entry): @@ -126,7 +162,7 @@ def del_cipher_pass(self, feature_type): Removes only the password for the given feature_type while keeping the file structure intact. """ try: - os.chmod(self._file_path, 0o777) + os.chmod(self._file_path, 0o640) with open(self._file_path, "r") as file: lines = file.readlines() @@ -145,3 +181,4 @@ def del_cipher_pass(self, feature_type): except Exception as e: syslog.syslog(syslog.LOG_ERR, "del_cipher_pass: {} Exception occurred: {}".format((e))) + From 33e88cb5b2c66363f1e29dc48a1e8ccd1b250006 Mon Sep 17 00:00:00 2001 From: nmoray Date: Mon, 19 May 2025 14:27:40 +0000 Subject: [PATCH 25/26] updated tests as per the new changes --- src/sonic-py-common/tests/test_security_cipher.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/sonic-py-common/tests/test_security_cipher.py b/src/sonic-py-common/tests/test_security_cipher.py index c3f169a873e3..2e7174825ce1 100644 --- a/src/sonic-py-common/tests/test_security_cipher.py +++ b/src/sonic-py-common/tests/test_security_cipher.py @@ -43,7 +43,7 @@ def test_passkey_encryption(self): mock_fd = mock.MagicMock() mock_fd.readlines = mock.MagicMock(return_value=DEFAULT_FILE) mock_file.return_value.__enter__.return_value = mock_fd - encrypt, err = temp.encrypt_passkey("TACPLUS", "passkey1", "TEST1") + encrypt = temp.encrypt_passkey("TACPLUS", "passkey1", "TEST1") assert encrypt != "passkey1" def test_passkey_decryption(self): @@ -58,7 +58,7 @@ def test_passkey_decryption(self): mock_fd = mock.MagicMock() mock_fd.readlines = mock.MagicMock(return_value=DEFAULT_FILE) mock_file.return_value.__enter__.return_value = mock_fd - encrypt, err = temp.encrypt_passkey("RADIUS", "passkey2", "TEST2") + encrypt = temp.encrypt_passkey("RADIUS", "passkey2", "TEST2") # Use patch to replace the built-in 'open' function with a mock #with mock.patch("{}.open".format(BUILTINS), mock.mock_open(read_data=EXPECTED_PASSWD)) as mock_file: @@ -67,7 +67,7 @@ def test_passkey_decryption(self): mock_fd = mock.MagicMock() mock_fd.readlines = mock.MagicMock(return_value=UPDATED_FILE) mock_file.return_value.__enter__.return_value = mock_fd - decrypt, err = temp.decrypt_passkey("RADIUS", encrypt) + decrypt = temp.decrypt_passkey("RADIUS", encrypt) assert decrypt == "passkey2" From f388f717f3c2f0e711b746a9b3a4a03c44dcce62 Mon Sep 17 00:00:00 2001 From: nmoray Date: Thu, 22 May 2025 09:15:31 +0000 Subject: [PATCH 26/26] Removed commented code --- src/sonic-py-common/tests/test_security_cipher.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/sonic-py-common/tests/test_security_cipher.py b/src/sonic-py-common/tests/test_security_cipher.py index 2e7174825ce1..792667c7eda9 100644 --- a/src/sonic-py-common/tests/test_security_cipher.py +++ b/src/sonic-py-common/tests/test_security_cipher.py @@ -61,7 +61,6 @@ def test_passkey_decryption(self): encrypt = temp.encrypt_passkey("RADIUS", "passkey2", "TEST2") # Use patch to replace the built-in 'open' function with a mock - #with mock.patch("{}.open".format(BUILTINS), mock.mock_open(read_data=EXPECTED_PASSWD)) as mock_file: with mock.patch("{}.open".format(BUILTINS), mock.mock_open()) as mock_file, \ mock.patch("os.chmod") as mock_chmod: mock_fd = mock.MagicMock()