Adding support for lockdown feature

io/pci/pci_hotplug.py

@@ -91,6 +91,17 @@ def setUp(self):
         if not os.path.isdir('/sys/bus/pci/slots/%s' % slot):
             self.cancel("%s is not present in sysfs path" % slot)
 
+        self.lockdown_mode = self.params.get("lockdown_mode", default="integrity")
+        self.lockdown_path = "/sys/kernel/security/lockdown"
+        self.lockdown_enable = self.params.get("lockdown_enable", default=False)
+        if self.lockdown_enable is True:
+            original_state = self.get_lockdown_state()
+            if original_state == "none":
+                if not self.set_lockdown_mode(self.lockdown_mode):
+                    self.fail(f"Failed to set lockdown to {self.lockdown_mode}")
+        else:
+            self.log.info("Running pci hotplug test case by default without setting lockdown feature")
+
     def test(self):
         """
         Removes and adds back a PCI adapter based on pci_adress.
@@ -239,3 +250,61 @@ def nvme_recovery_check():
                 return False
             return True
         return False
+
+    def check_lockdown_support(self):
+        '''
+        Check if kernel lockdown is supported
+        '''
+        if not os.path.exists(self.lockdown_path):
+            self.log.warn("Kernel lockdown not supported on this system")
+            return False
+        return True
+
+    def get_lockdown_state(self):
+        '''
+        Get current lockdown state
+        '''
+        try:
+            output = process.system_output(f'cat {self.lockdown_path}',
+                                           shell=True, sudo=True).decode("utf-8")
+            # Parse output like: "none [integrity] confidentiality"
+            if '[none]' in output:
+                return 'none'
+            elif '[integrity]' in output:
+                return 'integrity'
+            elif '[confidentiality]' in output:
+                return 'confidentiality'
+        except Exception as e:
+            self.log.error(f"Failed to get lockdown state: {e}")
+        return None
+
+    def set_lockdown_mode(self, mode):
+        '''
+        Set kernel lockdown mode
+        mode: 'none', 'integrity', or 'confidentiality'
+        '''
+        if not self.check_lockdown_support():
+            return False
+
+        current_state = self.get_lockdown_state()
+        self.log.info(f"Current lockdown state: {current_state}")
+
+        if mode == current_state:
+            self.log.info(f"Lockdown already set to {mode}")
+            return True
+
+        try:
+            cmd = f'echo "{mode}" > {self.lockdown_path}'
+            process.run(cmd, shell=True, sudo=True)
+
+            # Verify the change
+            new_state = self.get_lockdown_state()
+            if new_state == mode:
+                self.log.info(f"Successfully set lockdown to {mode}")
+                return True
+            else:
+                self.log.error(f"Failed to set lockdown to {mode}, current: {new_state}")
+                return False
+        except Exception as e:
+            self.log.error(f"Error setting lockdown mode: {e}")
+            return False

io/pci/pci_hotplug.py.data/pci_hotplug_with_lockdown.yaml

@@ -0,0 +1,7 @@
+pci_devices: ""
+count: 10
+#This option is for nvme splitter adapter, Value: nvme_splitter
+adapter_type: ""
+peer_ip:
+lockdown_mode: ""
+lockdown_enable: