Pankajjoshi/KEK URL rotation for CVM

VMEncryption/main/Common.py

@@ -60,9 +60,17 @@ class CommonVariables:
     """
     CVM LUKS2 header token
     """
+    #this token id is used to store Primary ADE (encryption setting + wrapped passphrase) token.
     cvm_ade_vm_encryption_token_id = 5
+    #this token id is used to store backup of token id 5, during CMK rotation.
+    cvm_ade_vm_encryption_backup_token_id = 6
     ADEEncryptionVersionInLuksToken_1_0='1.0'
-    PassphraseNameValue = 'LUKSPasswordProtector'
+    PassphraseNameValueProtected = 'LUKSPasswordProtector'
+    #this token type is used to store Primary ADE token. Type id: 5
+    AzureDiskEncryptionToken = 'Azure_Disk_Encryption'
+    #this token type is used to store backup ADE token. Type id: 6
+    #this token used for recovery to token 5 in case of reboot/interruption happened during KEK rotation.
+    AzureDiskEncryptionBackUpToken='Azure_Disk_Encryption_BackUp'
     """
     IMDS IP:
     """

VMEncryption/main/CryptMountConfigUtil.py

@@ -142,7 +142,7 @@ def _restore_backup_crypttab_info(self,crypt_item,passphrase_file):
             return False
 
         if not os.path.exists(azure_crypt_mount_backup_location) and not os.path.exists(crypttab_backup_location):
-            self.logger.log(msg=("MountPoint info not found for" + device_item_real_path), level=CommonVariables.ErrorLevel)
+            self.logger.log(msg=("MountPoint info not found for" + crypt_item.dev_path), level=CommonVariables.ErrorLevel)
             # Not sure when this happens..
             # in this case also, just add an entry to the azure_crypt_mount without a mount point.
             self.add_crypt_item(crypt_item)
@@ -265,19 +265,21 @@ def device_unlock_using_luks2_header(self):
         threads = []
         lock = threading.Lock()
         for device_item in device_items:
-            if device_item.file_system == "crypto_LUKS": 
+            if device_item.file_system == "crypto_LUKS":
+                 #restore 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))
                  threads.append(thread)
                  thread.start()
         for thread in threads:
-            thread.join()       
+            thread.join()
         self.logger.log("device_unlock_using_luks2_header End")
 
     def consolidate_azure_crypt_mount(self, passphrase_file):
         """
-        Reads the backup files from block devices that have a LUKS header and adds it to the cenral azure_crypt_mount file
+        Reads the backup files from block devices that have a LUKS2 header and adds it to the central azure_crypt_mount file
         """
         self.logger.log("Consolidating azure_crypt_mount")
 

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,60 @@ 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.
+        token data is as follow for version 1.0.
+        "version": "1.0",
+        "type": "Azure_Disk_Encryption",
+        "keyslots": [],
+        "KekVaultResourceId": "<kek_res_id>",
+        "KeyEncryptionKeyURL": "<kek_url>",
+        "KeyVaultResourceId": "<kv_res_id>",
+        "KeyVaultURL": "https://<vault_name>.vault.azure.net/",
+        "AttestationURL": null,
+        "PassphraseName": "LUKSPasswordProtector",
+        "Passphrase": "M53XE09n7O9r2AdKa7FYRYe..."
+        '''
+        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,41 +259,66 @@ 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)
-        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)
+        status = self.import_token_data(device_path=device_path,
+                                        token_data=data,
+                                        token_id=CommonVariables.cvm_ade_vm_encryption_token_id)
         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))
-        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)
+        return status
+
+    def read_token(self,device_name,token_id):
+        '''this functions reads tokens from LUKS2 header.'''
+        device_path = self.get_device_path(dev_name=device_name)
+        if not 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("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(level=CommonVariables.WarningLevel,
+                            msg="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
-        disk_encryption_setting=json.loads(token)
+        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(level=CommonVariables.WarningLevel,
+                            msg="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))
+        device_path = self.get_device_path(device_name)
+        if not device_path:
+            self.logger.log(level= CommonVariables.WarningLevel, msg="export_token Input is not valid. device name: {0}".format(device_name))
+            return None
+        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
+        disk_encryption_setting=self.read_token(device_name=device_name,token_id=cvm_ade_vm_encryption_token_id)
         if disk_encryption_setting['version'] != CommonVariables.ADEEncryptionVersionInLuksToken_1_0:
             self.logger.log("export_token token version {0} is not a vaild version.".format(disk_encryption_setting['version']))
             return None
         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(level=CommonVariables.WarningLevel, msg="passphrase is not Protected. 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 +462,50 @@ 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 len(token) == 2 and token[1] == token_name:
+                return token[0]
+        return None
+
+    def restore_luks2_token(self, device_name=None):
+        '''this function restores token
+        type:Azure_Disk_Encryption_BackUp, id:6 to type:Azure_Disk_Encryption id:5,
+        this function acts on 4 scenarios.
+        1. both token id: 5 and 6 present in LUKS2 Tokens filed, due to reboot/interrupt during
+        KEK rotation, such case remove token id 5 has latest data so remove token id 6.
+        2. token id 5 present but 6 is not present in LUKS2 Tokens field. do nothing.
+        3. token id 5 not present but 6 present in LUKS2 Tokens field, restore token id 5 using
+        token id 6, then remove token id 6.
+        4. no token ids 5 or 6 present in LUKS2 Tokens field, do nothing.'''
+        device_path = self.get_device_path(device_name)
+        if not device_path:
+            self.logger.log(level=CommonVariables.WarningLevel,msg="restore_luks2_token invalid input. device_name = {0}".format(device_name))
+            return
+        ade_token_id_primary = 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_primary:
+            #remove backup token id
+            self.remove_token(device_name=device_name,token_id=ade_token_id_backup)
+            return
+        self.logger.log("restore 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.AzureDiskEncryptionToken
+        self.import_token_data(device_path=device_path,token_data=data,token_id=ade_token_id_primary)
+        #remove backup
+        self.remove_token(device_name=device_name,token_id=ade_token_id_backup)
+        self.logger.log("restore luks2 token id {0} to {1} for device {2} is successful.".format(ade_token_id_backup,ade_token_id_primary,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 +519,47 @@ 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):
+        """
+        A luks v2 luksheader looks kind of like this: (inessential stuff removed)
+
+        LUKS header information
+        Version:        2
+        Data segments:
+            0: crypt
+                offset: 0 [bytes]
+                length: 5539430400 [bytes]
+                cipher: aes-xts-plain64
+                sector: 512 [bytes]
+        Keyslots:
+            1: luks2
+                    Key:        512 bits
+            3: reencrypt (unbound)
+                    Key:        8 bits
+        Tokens:
+            6: 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,14 @@ def _is_kv_equivalent(self, a, b):
             if b[-1] == '/': b = b[:-1]
         return a==b
 
+    def cmk_changed(self):
+        '''current config CMK changed from effective config CMK.'''
+        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 \