Adding support of common security cipher module for encryption and decryption of a passkey

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

@@ -0,0 +1,118 @@
+'''
+
+A common module for handling the encryption and
+decryption of the feature passkey. It also takes
+care of storing 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()
+    _initialized = False
+
+    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")
+            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'
+
+                    with open(self._file_path, 'w') as file:
+                        file.writelines(lines)
+            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
+

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):
+        pass
+
+    def get(self, db, table, field):
+        return self.data.get(field, "N/A")

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

@@ -0,0 +1,68 @@
+import sys
+
+if sys.version_info.major == 3:
+    from unittest import mock
+else:
+    import mock
+
+import pytest
+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__"
+
+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):
+        with mock.patch("sonic_py_common.security_cipher.ConfigDBConnector", new=ConfigDBConnector), \
+             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:
+                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"
+
+    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:
+            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:
+                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")
+
+            # 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_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 decrypt == "passkey2"
+
+

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 ",")';
                         }