Pankajjoshi/KEK URL rotation for CVM

VMEncryption/main/Common.py

@@ -60,9 +60,15 @@ class CommonVariables:
     """
     CVM LUKS2 header token
     """
+    #used to store ADE (encryption setting + wrapped passphrase) token. Type: Azure_Disk_Encryption 
     cvm_ade_vm_encryption_token_id = 5
+    #used to store backup of token type Azure_Disk_Encryption token during CMK rotation. Type: Azure_Disk_Encryption_BackUp
+    cvm_ade_vm_encryption_backup_token_id = 6
     ADEEncryptionVersionInLuksToken_1_0='1.0'
-    PassphraseNameValue = 'LUKSPasswordProtector'
+    PassphraseNameValueProtected = 'LUKSPasswordProtector'
+    PassphraseNameValueNotProtected = 'LUKSPasswordNotProtector'
+    AzureDiskEncryptionToken = 'Azure_Disk_Encryption'
+    AzureDiskEncryptionBackUpToken='Azure_Disk_Encryption_BackUp'
     """
     IMDS IP:
     """

VMEncryption/main/CryptMountConfigUtil.py

@@ -266,6 +266,8 @@ def device_unlock_using_luks2_header(self):
         lock = threading.Lock()
         for device_item in device_items:
             if device_item.file_system == "crypto_LUKS": 
+                 #resotre Luks2 token using BackUp.
+                 self.disk_util.restore_luks2_token(device_name=device_item.name)
                  device_item_path = self.disk_util.get_device_path(device_item.name)
                  azure_item_path = azure_name_table[device_item_path] if device_item_path in azure_name_table else device_item_path
                  thread = threading.Thread(target=self._device_unlock_using_luks2_header,args=(device_item.name,device_item_path,azure_item_path,lock))

VMEncryption/main/DiskUtil.py

@@ -24,6 +24,7 @@
 from subprocess import Popen
 import traceback
 import glob
+import tempfile
 
 from EncryptionConfig import EncryptionConfig
 from DecryptionMarkConfig import DecryptionMarkConfig
@@ -193,21 +194,48 @@ def secure_key_release_operation(self,protectorbase64,kekUrl,operation,attestati
         self.logger.log("secure_key_release_operation {0} end.".format(operation))
         return process_comm.stdout.strip()
 
-    def import_token(self,device_path,passphrase_file,public_settings):
+    def import_token_data(self,device_path,token_data,token_id):
+        '''Updating token_data json object to LUKS2 header's Tokens field.'''
+        self.logger.log(msg="import_token_data for device: {0} started.".format(device_path))
+        if not token_data or not type(token_data) is dict:
+            self.logger.log(level=CommonVariables.WarningLevel, msg="import_token_data: token_data: {0} for device: {1} is not valid.".format(token_data,device_path))
+            return False
+        if not token_id:
+            self.logger.log(level= CommonVariables.WarningLevel, msg = "import_token_data: token_id: {0} for device: {1} is not valid.".format(token_id,device_path) )
+            return False
+        temp_file = tempfile.NamedTemporaryFile(delete=False,mode='w+')
+        json.dump(token_data,temp_file,indent=4)
+        temp_file.close()
+        cmd = "cryptsetup token import --json-file {0} --token-id {1} {2}".format(temp_file.name,token_id,device_path)
+        process_comm = ProcessCommunicator()
+        status = self.command_executor.Execute(cmd,communicator=process_comm)
+        self.logger.log(msg="import_token_data: device: {0} status: {1}".format(device_path,status))
+        os.unlink(temp_file.name)
+        return status==CommonVariables.process_success
+
+    def import_token(self,device_path,passphrase_file,public_settings,PassphraseNameValue=CommonVariables.PassphraseNameValueProtected):
         '''this function reads passphrase from passphrase file, wrap it and update in token field of LUKS2 header.'''
         self.logger.log(msg="import_token for device: {0} started.".format(device_path))
-        protector = ""
+        self.logger.log(msg="import_token for passphrase file path: {0}.".format(passphrase_file))
+        if not passphrase_file or not os.path.exists(passphrase_file):
+            self.logger.log(level=CommonVariables.WarningLevel,msg="import_token for passphrase file path: {0} not exists.".format(passphrase_file))
+            return False
+        Protector= ""
         with open(passphrase_file,"rb") as protector_file:
             #passphrase stored in keyfile is base64
-            protector = protector_file.read().decode('utf-8')
+            Protector = protector_file.read().decode('utf-8')
         KekVaultResourceId=public_settings.get(CommonVariables.KekVaultResourceIdKey)
         KeyEncryptionKeyUrl=public_settings.get(CommonVariables.KeyEncryptionKeyURLKey)
         AttestationUrl = public_settings.get(CommonVariables.AttestationURLKey)
-        wrappedProtector = self.secure_key_release_operation(protectorbase64=protector,
+        if PassphraseNameValue == CommonVariables.PassphraseNameValueProtected:
+            Protector = self.secure_key_release_operation(protectorbase64=Protector,
                                                         kekUrl=KeyEncryptionKeyUrl,
                                                         operation=CommonVariables.secure_key_release_wrap,
                                                         attestationUrl=AttestationUrl)
-        if not wrappedProtector:
+        else:
+            self.logger.log(msg="import_token passphrase is not wrapped, value of passphrase name key: {0}".format(PassphraseNameValue))
+
+        if not Protector:
             self.logger.log("import_token protector wrapping is unsuccessful for device {0}".format(device_path))
             return False
         data={
@@ -219,33 +247,61 @@ def import_token(self,device_path,passphrase_file,public_settings):
             CommonVariables.KeyVaultResourceIdKey:public_settings.get(CommonVariables.KeyVaultResourceIdKey),
             CommonVariables.KeyVaultURLKey:public_settings.get(CommonVariables.KeyVaultURLKey),
             CommonVariables.AttestationURLKey:AttestationUrl,
-            CommonVariables.PassphraseNameKey:CommonVariables.PassphraseNameValue,
-            CommonVariables.PassphraseKey:wrappedProtector
+            CommonVariables.PassphraseNameKey:PassphraseNameValue,
+            CommonVariables.PassphraseKey:Protector
         }
-        #TODO: needed to decide on temp path.
-        custom_cmk = os.path.join("/var/lib/azure_disk_encryption_config/","custom_cmk.json")
-        out_file = open(custom_cmk,"w")
-        json.dump(data,out_file,indent=4)
-        out_file.close()
-        cmd = "cryptsetup token import --json-file {0} --token-id {1} {2}".format(custom_cmk,CommonVariables.cvm_ade_vm_encryption_token_id,device_path)
+        temp_file = tempfile.NamedTemporaryFile(delete=False,mode='w+')
+        json.dump(data,temp_file,indent=4)
+        temp_file.close()
+        cmd = "cryptsetup token import --json-file {0} --token-id {1} {2}".format(temp_file.name,CommonVariables.cvm_ade_vm_encryption_token_id,device_path)
         process_comm = ProcessCommunicator()
         status = self.command_executor.Execute(cmd,communicator=process_comm)
         self.logger.log(msg="import_token: device: {0} status: {1}".format(device_path,status))
-        os.remove(custom_cmk)
+        os.unlink(temp_file.name)
         self.logger.log(msg="import_token: device: {0} end.".format(device_path))
         return status==CommonVariables.process_success
 
-    def export_token(self,device_name):
-        '''This function reads token id from luks2 header field and unwrap passphrase'''
-        self.logger.log("export_token to device {0} started.".format(device_name))
+    def read_token(self,device_name,token_id):
+        '''this functions reads tokens from LUKS2 header.'''
         device_path = os.path.join("/dev",device_name)
-        protector = None
-        cmd = "cryptsetup token export --token-id {0} {1}".format(CommonVariables.cvm_ade_vm_encryption_token_id,device_path)
+        if not os.path.exists(device_path) or not token_id:
+            self.logger.log(level=CommonVariables.WarningLevel,msg="read_token: Inputs not valid. device_name: {0}, token id: {1}".format(device_name,token_id))
+            return None
+        cmd = "cryptsetup token export --token-id {0} {1}".format(token_id,device_path)
+        process_comm = ProcessCommunicator()
+        status = self.command_executor.Execute(cmd, communicator=process_comm)
+        if status != 0:
+            self.logger.log("read_token token id {0} not found in device {1} LUKS header".format(CommonVariables.cvm_ade_vm_encryption_token_id,device_name))
+            return None
+        token = process_comm.stdout
+        return json.loads(token)
+
+    def remove_token(self,device_name,token_id):
+        '''this function remove the token'''
+        device_path = os.path.join("/dev",device_name)
+        cmd = "cryptsetup token remove --token-id {0} {1}".format(token_id,device_path)
         process_comm = ProcessCommunicator()
         status = self.command_executor.Execute(cmd, communicator=process_comm)
         if status != 0:
-            self.logger.log("export_token token id {0} not found in device {1} LUKS header".format(CommonVariables.cvm_ade_vm_encryption_token_id,device_name))
+            self.logger.log("remove token id {0} not found in device {1} LUKS header".format(token_id,device_name))
+            return False
+        return True
+
+    def export_token(self,device_name):
+        '''This function reads token id from luks2 header field and unwrap passphrase'''
+        self.logger.log("export_token to device {0} started.".format(device_name))
+        if not device_name or not os.path.exists(self.get_device_path(device_name)):
+            self.logger.log(level= CommonVariables.WarningLevel, msg="export_token Input is not valid. device name: {0}".format(device_name))
             return None
+        device_path = self.get_device_path(device_name)
+        protector = None
+        cvm_ade_vm_encryption_token_id = self.get_token_id(header_or_dev_path=device_path,token_name=CommonVariables.AzureDiskEncryptionToken)
+        if not cvm_ade_vm_encryption_token_id:
+            self.logger.log("export_token token id {0} not found in device {1} LUKS header".format(cvm_ade_vm_encryption_token_id,device_name))
+            return None
+        cmd = "cryptsetup token export --token-id {0} {1}".format(cvm_ade_vm_encryption_token_id,device_path)
+        process_comm = ProcessCommunicator()
+        self.command_executor.Execute(cmd, communicator=process_comm)
         token = process_comm.stdout
         disk_encryption_setting=json.loads(token)
         if disk_encryption_setting['version'] != CommonVariables.ADEEncryptionVersionInLuksToken_1_0:
@@ -254,6 +310,9 @@ def export_token(self,device_name):
         keyEncryptionKeyUrl=disk_encryption_setting[CommonVariables.KeyEncryptionKeyURLKey]
         wrappedProtector = disk_encryption_setting[CommonVariables.PassphraseKey]
         attestationUrl = disk_encryption_setting[CommonVariables.AttestationURLKey]
+        if disk_encryption_setting[CommonVariables.PassphraseNameKey] != CommonVariables.PassphraseNameValueProtected:
+            self.logger.log("passphrase is not Protectected. No need to do SKR.")
+            return wrappedProtector if wrappedProtector else None
         if wrappedProtector:
             #unwrap the protector.
             protector=self.secure_key_release_operation(attestationUrl=attestationUrl,
@@ -397,6 +456,47 @@ def luks_get_uuid(self, header_or_dev_path):
                 return splits[1]
         return None
 
+    def get_token_id(self,header_or_dev_path,token_name):
+        '''if LUKS2 header has token name return the id else return none.'''
+        if not header_or_dev_path or not os.path.exists(header_or_dev_path) or not token_name:
+            self.logger.log("get_token_id: invalid input, header_or_dev_path:{0} token_name:{1}".format(header_or_dev_path,token_name))
+            return None
+        luks_dump_out = self._luks_get_header_dump(header_or_dev_path)
+        tokens = self._extract_luksv2_token(luks_dump_out)
+        for token in tokens:
+            if token[1] == token_name:
+                return token[0]
+        return None
+
+    def restore_luks2_token(self, device_name=None):
+        '''this function restoring token type Azure_Disk_Encryption_BackUp to Azure_Disk_Encryption, 
+        this function acts on 4 secenarios. [token id Azure_Disk_Encryption, token id Azure_Disk_Encryption_BackUp, resote_action)'''
+        '''[Y,Y,remove token type Azure_Disk_Encryption_BackUp]
+           [Y,N,do nothing]
+           [N,N,do nothing]
+           [N,Y,move token type Azure_Disk_Encryption_BackUp to Azure_Disk_Encryption]'''
+        if not device_name or not os.path.exists(self.get_device_path(device_name)):
+            self.logger.log(level=CommonVariables.WarningLevel,msg="restore_luks2_token invalid input. device_name = {0}".format(device_name))
+            return
+        device_path = self.get_device_path(device_name)
+        ade_token_id = self.get_token_id(header_or_dev_path=device_path,token_name=CommonVariables.AzureDiskEncryptionToken)
+        ade_token_id_backup = self.get_token_id(header_or_dev_path=device_path,token_name=CommonVariables.AzureDiskEncryptionBackUpToken)
+        if not ade_token_id_backup:
+            #do nothing
+            return
+        if ade_token_id:
+            #remove backup token id
+            self.remove_token(device_name=device_name,token_id=ade_token_id_backup)
+            return
+        self.logger.log("resotre luks2 token for device {0} is started.".format(device_name))
+        #read from backup and update AzureDiskEncryptionToken
+        data = self.read_token(device_name=device_name,token_id=ade_token_id_backup)
+        data['type']=CommonVariables.AzureDiskEncryptionBackUpToken
+        self.import_token_data(device_path=device_path,token_data=data,token_id=CommonVariables.AzureDiskEncryptionToken)
+        #remove backup
+        self.remove_token(device_name=device_name,token_id=ade_token_id_backup)
+        self.logger.log("resotre luks2 token for device {0} is successful.".format(device_name))
+
     def _get_cryptsetup_version(self):
         # get version of currently installed cryptsetup
         cryptsetup_cmd = "{0} --version".format(self.distro_patcher.cryptsetup_path)
@@ -410,6 +510,33 @@ def _extract_luks_version_from_dump(self, luks_dump_out):
             if "version:" in line.lower():
                 return line.split()[-1]
 
+    def _extract_luksv2_token(self, luks_dump_out):
+        """
+        ...
+        Tokens:
+            1: Azure_Disk_Encryption_BackUp
+            5: Azure_Disk_Encryption
+        ...
+        """
+        if not luks_dump_out:
+            return []
+        lines = luks_dump_out.split("\n")
+        token_segment = False
+        token_lines = []
+        for line in lines:
+            parts = line.split(":")
+            if len(parts)<2:
+                continue
+            if token_segment and parts[1].strip() == '':
+                break
+            if "tokens" in parts[0].strip().lower():
+                token_segment = True
+                continue
+            if token_segment and self._isnumeric(parts[0].strip()):
+                token_lines.append([int(parts[0].strip()),parts[1].strip()])
+                continue
+        return token_lines
+
     def _extract_luksv2_keyslot_lines(self, luks_dump_out):
         """
         A luks v2 luksheader looks kind of like this: (inessential stuff removed)

VMEncryption/main/ExtensionParameter.py

@@ -174,6 +174,13 @@ def _is_kv_equivalent(self, a, b):
             if b[-1] == '/': b = b[:-1]
         return a==b
 
+    def cmk_changed(self):
+        if (self.KeyEncryptionKeyURL or self.get_kek_url()) and \
+           (not self._is_kv_equivalent(self.KeyEncryptionKeyURL, self.get_kek_url())):
+           self.logger.log('Current config KeyEncryptionKeyURL {0} differs from effective config KeyEncryptionKeyURL {1}'.format(self.KeyEncryptionKeyURL, self.get_kek_url()))
+           return True
+        return False
+
     def config_changed(self):
         if (self.command or self.get_command()) and \
            (self.command != self.get_command() and \