Pankajjoshi/KEK URL rotation for CVM

VMEncryption/main/Common.py

@@ -61,8 +61,12 @@ class CommonVariables:
     CVM LUKS2 header token
     """
     cvm_ade_vm_encryption_token_id = 5
+    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 from BackUp (if needed)
+                 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,37 @@ 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):
+        self.logger.log(msg="import_token for device: {0} started.".format(device_path))
+        temp_file = tempfile.NamedTemporaryFile(delete=False)
+        temp_file.close()
+        json.dump(token_data,temp_file.name,indent=4)
+        cmd = "cryptsetup token import --json-file {0} --token-id {1} {2}".format(temp_file.name,token_id)
+        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.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 = ""
+        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,10 +236,10 @@ 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.
+        #TODO handle with temp file.
         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)
@@ -235,10 +252,33 @@ def import_token(self,device_path,passphrase_file,public_settings):
         self.logger.log(msg="import_token: device: {0} end.".format(device_path))
         return status==CommonVariables.process_success
 
+    def read_token(self,device_name,token_id):
+        '''this functions reads tokens from LUKS2 header.'''
+        device_path = os.path.join("/dev",device_name)
+        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))
+            return None
+        token = process_comm.stdout
+        return 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("remove token id {0} not found in device {1} LUKS header".format(CommonVariables.cvm_ade_vm_encryption_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 = os.path.join("/dev",device_name)
+        device_path = self.get_device_path(device_name)
         protector = None
         cmd = "cryptsetup token export --token-id {0} {1}".format(CommonVariables.cvm_ade_vm_encryption_token_id,device_path)
         process_comm = ProcessCommunicator()
@@ -254,6 +294,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 +440,38 @@ 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.'''
+        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] is 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'''
+        if not 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 +485,31 @@ 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
+        ...
+        """
+        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() is '':
+                break
+            if "tokens" in parts[0].strip().lower():
+                token_segment = True
+                continue
+            if token_segment and self._isnumeric(parts[0].strip()):
+                token_lines.append(parts)
+                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 \

VMEncryption/main/handle.py

@@ -185,15 +185,16 @@ def disable_encryption():
 
 
 def stamp_disks_with_settings(items_to_encrypt, encryption_config, encryption_marker=None):
-    if security_Type == CommonVariables.ConfidentialVM:
-        logger.log(msg="Do not send vm setting to host for stamping.",level=CommonVariables.InfoLevel)
-        return
     disk_util = DiskUtil(hutil=hutil, patching=DistroPatcher, logger=logger, encryption_environment=encryption_environment)
     crypt_mount_config_util = CryptMountConfigUtil(logger=logger, encryption_environment=encryption_environment, disk_util=disk_util)
     bek_util = BekUtil(disk_util, logger,encryption_environment)
     current_passphrase_file = bek_util.get_bek_passphrase_file(encryption_config)
     public_settings = get_public_settings()
     extension_parameter = ExtensionParameter(hutil, logger, DistroPatcher, encryption_environment, get_protected_settings(), public_settings)
+    if security_Type == CommonVariables.ConfidentialVM:
+        logger.log(msg="Do not send vm setting to host for stamping.",level=CommonVariables.InfoLevel)
+        extension_parameter.commit()
+        return
     has_keystore_flag = CommonVariables.KeyStoreTypeKey in public_settings
 
     # post new encryption settings via wire server protocol
@@ -270,6 +271,77 @@ def get_protected_settings():
     else:
         return protected_settings_str
 
+def update_encryption_settings_luks2_header():
+    '''This function is used for CMK passphrse wrapping with new KEK URL and update metadata in LUKS2 header.'''
+    hutil.do_parse_context('UpdateEncryptionSettingsLuks2Header')
+    logger.log('Updating encryption settings LUKS-2 header')
+    # ensure cryptsetup package is still available in case it was for some reason removed after enable
+    try:
+        DistroPatcher.install_cryptsetup()
+    except Exception as e:
+        hutil.save_seq()
+        message = "Failed to update encryption settings with error: {0}, stack trace: {1}".format(e, traceback.format_exc())
+        hutil.do_exit(exit_code=CommonVariables.missing_dependency,
+                      operation='UpdateEncryptionSettingsLuks2Header',
+                      status=CommonVariables.extension_error_status,
+                      code=str(CommonVariables.missing_dependency),
+                      message=message)
+    try:
+        public_setting = get_public_settings()
+        encryption_config = EncryptionConfig(encryption_environment, logger)
+        extension_parameter = ExtensionParameter(hutil, logger, DistroPatcher, encryption_environment, get_protected_settings(), public_setting)
+        disk_util = DiskUtil(hutil=hutil, patching=DistroPatcher, logger=logger, encryption_environment=encryption_environment)
+        bek_util = BekUtil(disk_util, logger,encryption_environment)
+        device_items = disk_util.get_device_items(None)
+        for device_item in device_items:
+            device_item_path = disk_util.get_device_path(device_item.name)
+            if not disk_util.is_luks_device(device_item_path,None):
+                logger.log("Not a LUKS device, device path: {0}".format(device_item_path))
+                continue
+            #estoring the token data to type Azure_Disk_Encryption
+            disk_util.restore_luks2_token(device_name=device_item.name)
+            logger.log("Reading passphrase from LUKS2 header, device name: {0}".format(device_item.name))
+            #keep token copy for manual recovery
+            ade_token_id = disk_util.get_token_id(header_or_dev_path=device_item_path)
+            if not ade_token_id:
+                logger.log("token type: Azure_Disk_Encryption not found for device {0}".format(device_item.name))
+                continue
+            data = disk_util.read_token(device_name=device_item.name,token_id=ade_token_id)
+            #writing to backup token
+            data['type']=CommonVariables.AzureDiskEncryptionBackUpToken
+            disk_util.import_token_data(device_path=device_item_path,token_data=data,token_id=CommonVariables.cvm_ade_vm_encryption_backup_token_id)
+            #get the unwrapped passphrase from LUKS2 header.
+            passphrase=disk_util.export_token(device_name=device_item.name)
+            #remove token
+            disk_util.remove_token(device_name=device_item.name,token_id=ade_token_id)
+            if not passphrase:
+                logger.log("No passphrase in LUKS2 header, device name: {0}".format(device_item.name))
+                continue
+            logger.log("Updating wrapped passphrase to LUKS2 header with current public setting. device name {0}".format(device_item.name))
+            #protect passphrase before updating to LUKS2 is done in import_token
+            temp_keyfile = tempfile.NamedTemporaryFile(delete=False)
+            temp_keyfile.write(passphrase.encode("utf-8"))
+            temp_keyfile.close()
+            #save passphrase to LUKS2 header with PassphraseNameValueProtected
+            ret = disk_util.import_token(device_path=device_item_path,passphrase_file=temp_keyfile.name,public_settings=public_setting,PassphraseNameValue=CommonVariables.PassphraseNameValueProtected)
+            if not ret:
+                logger.log("Update passphrase with current public setting to LUKS2 header is not successful. device path {0}".format(device_item_path))
+                return None
+            os.unlink(temp_keyfile.name)
+            #removing backup token
+            disk_util.remove_token(device_name=device_item.name,token_id=CommonVariables.cvm_ade_vm_encryption_backup_token_id)
+        extension_parameter.commit()
+        bek_util.umount_azure_passhprase(encryption_config)
+    except Exception as e:
+        hutil.save_seq()
+        message = "Failed to update encryption settings Luks2 header with error: {0}, stack trace: {1}".format(e, traceback.format_exc())
+        logger.log(msg=message, level=CommonVariables.ErrorLevel)
+        bek_util.umount_azure_passhprase(encryption_config)
+        hutil.do_exit(exit_code=CommonVariables.unknown_error,
+                      operation='UpdateEncryptionSettingsLuks2Header',
+                      status=CommonVariables.extension_error_status,
+                      code=str(CommonVariables.unknown_error),
+                      message=message)
 
 def update_encryption_settings(extra_items_to_encrypt=[]):
     hutil.do_parse_context('UpdateEncryptionSettings')
@@ -945,14 +1017,21 @@ def handle_encryption(public_settings, encryption_status, disk_util, bek_util, e
             logger.log("An operation already running. Cannot accept an update settings request.")
             hutil.reject_settings()
         are_devices_encrypted, items_to_encrypt = are_required_devices_encrypted(volume_type, encryption_status, disk_util, bek_util, encryption_operation)
-        if not are_devices_encrypted:
-            logger.log('Required devices not encrypted for volume type {0}. Calling update to stamp encryption settings.'.format(volume_type))
-            update_encryption_settings(items_to_encrypt)
-            logger.log('Encryption Settings stamped. Calling enable to encrypt new devices.')
-            enable_encryption()
+        if security_Type==CommonVariables.ConfidentialVM:
+            logger.log('Calling Update Encryption Setting in LUKS2 header.')
+            if extension_parameter.cmk_changed():
+                update_encryption_settings_luks2_header()
+            if not are_devices_encrypted:
+                enable_encryption()
         else:
-            logger.log('Calling Update Encryption Setting.')
-            update_encryption_settings()
+            if not are_devices_encrypted:
+                logger.log('Required devices not encrypted for volume type {0}. Calling update to stamp encryption settings.'.format(volume_type))
+                update_encryption_settings(items_to_encrypt)
+                logger.log('Encryption Settings stamped. Calling enable to encrypt new devices.')
+                enable_encryption()
+            else:
+                logger.log('Calling Update Encryption Setting.')
+                update_encryption_settings()
     else:
         logger.log("Config did not change or first call, enabling encryption")
         encryption_marker = EncryptionMarkConfig(logger, encryption_environment)