smm_ptr: Add SMI contention check to the scan mode

chipsec/modules/tools/smm/smm_ptr.py

@@ -81,6 +81,7 @@
 
 import struct
 import os
+import sys
 
 from chipsec.module_common import BaseModule
 from chipsec.library.returncode import ModuleResult
@@ -146,6 +147,9 @@
 # be long-running
 OUTLIER_THRESHOLD = 33.3
 
+# SMI count MSR
+MSR_SMI_COUNT = 0x00000034
+
 #
 # Defaults
 #
@@ -195,6 +199,28 @@ def __init__(self):
         self.hist_smi_duration = 0
         self.hist_smi_num = 0
         self.outliers_hist = 0
+        self.smi_count = self.get_smi_count()
+
+    def get_smi_count(self):
+        #
+        # The SMI count is the same on all CPUs
+        #
+        cpu = 0
+        fd = os.open(f"/dev/cpu/{cpu}/msr", os.O_RDONLY)
+        os.lseek(fd, MSR_SMI_COUNT, os.SEEK_SET)
+        count = struct.unpack('Q', os.read(fd, 8))[0]
+        os.close(fd)
+        return count
+
+    def valid_smi_count(self):
+        valid = False
+        count = self.get_smi_count()
+        if (count == self.smi_count + 1):
+            valid = True
+        self.smi_count = count
+        if not valid:
+            print("SMI contention detected", file=sys.stderr)
+        return valid
 
     def find_address_in_regs(self, gprs):
         for key, value in gprs.items():
@@ -204,9 +230,9 @@ def find_address_in_regs(self, gprs):
                 return key
 
     def clear(self):
-        self.max = smi_info(0);
-        self.min = smi_info(2**32-1);
-        self.outlier = smi_info(0);
+        self.max = smi_info(0)
+        self.min = smi_info(2**32-1)
+        self.outlier = smi_info(0)
         self.acc_smi_duration = 0
         self.acc_smi_num = 0
         self.avg_smi_duration = 0
@@ -360,7 +386,7 @@ def send_smi_timed(self, thread_id, smi_code, smi_data, name, desc, rax, rbx, rc
             self.logger.log(f'      RSI: 0x{rsi:016X}')
             self.logger.log(f'      RDI: 0x{rdi:016X}')
         ret = self.interrupts.send_SW_SMI_timed(thread_id, smi_code, smi_data, rax, rbx, rcx, rdx, rsi, rdi)
-        duration = ret[7];
+        duration = ret[7]
         return (True, duration)
 
     def check_memory(self, _addr, _smi_desc, fn, restore_contents=False):
@@ -443,12 +469,18 @@ def smi_fuzz_iter(self, thread_id, _addr, _smi_desc, fill_contents=True, restore
         if not scan:
             self.send_smi(thread_id, _smi_desc.smi_code, _smi_desc.smi_data, _smi_desc.name, _smi_desc.desc, _rax, _rbx, _rcx, _rdx, _rsi, _rdi)
         else:
-            _, duration = self.send_smi_timed(thread_id, _smi_desc.smi_code, _smi_desc.smi_data, _smi_desc.name, _smi_desc.desc, _rax, _rbx, _rcx, _rdx, _rsi, _rdi)
+            while True:
+                _, duration = self.send_smi_timed(thread_id, _smi_desc.smi_code, _smi_desc.smi_data, _smi_desc.name, _smi_desc.desc, _rax, _rbx, _rcx, _rdx, _rsi, _rdi)
+                if scan.valid_smi_count():
+                    break
             #
             # Re-do the call if it was identified as an outlier, due to periodic SMI delays
             #
             if scan.is_outlier(duration):
-                _, duration = self.send_smi_timed(thread_id, _smi_desc.smi_code, _smi_desc.smi_data, _smi_desc.name, _smi_desc.desc, _rax, _rbx, _rcx, _rdx, _rsi, _rdi)
+                while True:
+                    _, duration = self.send_smi_timed(thread_id, _smi_desc.smi_code, _smi_desc.smi_data, _smi_desc.name, _smi_desc.desc, _rax, _rbx, _rcx, _rdx, _rsi, _rdi)
+                    if scan.valid_smi_count():
+                        break
         #
         # Check memory buffer if not in 'No Fill' mode
         #