mirror of
https://github.com/chipsec/chipsec
synced 2026-06-08 13:31:00 +00:00
William Leara
79f81bbb7e
This commit fixes whitespace errors from checkers W291,W292,W293,W391. They were found by running "flake8 ." from the root directory. These are errors defined by "pycodestyle", and have no functional impact to the source. This is simply fixing whitespace as defined by CHIPSEC's .flake8 configuration. background: https://www.flake8rules.com/rules/W291.html https://www.flake8rules.com/rules/W292.html https://www.flake8rules.com/rules/W293.html https://www.flake8rules.com/rules/W391.html tool versions: flake8 v7.3.0, python v3.13.11 Signed-off-by: William Leara <william.leara@dell.com>
241 lines
10 KiB
Python
241 lines
10 KiB
Python
# CHIPSEC: Platform Security Assessment Framework
|
|
# Copyright (c) 2010-2021, Intel Corporation
|
|
#
|
|
# This program is free software; you can redistribute it and/or
|
|
# modify it under the terms of the GNU General Public License
|
|
# as published by the Free Software Foundation; Version 2.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program; if not, write to the Free Software
|
|
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
|
|
#
|
|
# Contact information:
|
|
# chipsec@intel.com
|
|
#
|
|
|
|
|
|
"""
|
|
CPU related functionality
|
|
|
|
"""
|
|
from typing import Dict, List, Tuple, Optional
|
|
from chipsec.hal import hal_base
|
|
from chipsec.hal.common import acpi
|
|
from chipsec.library.logger import logger
|
|
from chipsec.library import paging
|
|
from chipsec.library.cpu import VMM_NONE, VMM_XEN, VMM_HYPER_V, VMM_VMWARE, VMM_KVM
|
|
|
|
|
|
|
|
########################################################################################################
|
|
#
|
|
# CORES HAL Component
|
|
#
|
|
########################################################################################################
|
|
|
|
class CPU(hal_base.HALBase):
|
|
def __init__(self, cs):
|
|
super(CPU, self).__init__(cs)
|
|
self.helper = cs.helper
|
|
|
|
def read_cr(self, cpu_thread_id: int, cr_number: int) -> int:
|
|
value = self.helper.read_cr(cpu_thread_id, cr_number)
|
|
logger().log_hal(f'[cpu{cpu_thread_id:d}] read CR{cr_number:d}: value = 0x{value:08X}')
|
|
return value
|
|
|
|
def write_cr(self, cpu_thread_id: int, cr_number: int, value: int) -> int:
|
|
logger().log_hal(f'[cpu{cpu_thread_id:d}] write CR{cr_number:d}: value = 0x{value:08X}')
|
|
status = self.helper.write_cr(cpu_thread_id, cr_number, value)
|
|
return status
|
|
|
|
def cpuid(self, eax: int, ecx: int) -> Tuple[int, int, int, int]:
|
|
logger().log_hal(f'[cpu] CPUID in : EAX=0x{eax:08X}, ECX=0x{ecx:08X}')
|
|
(eax, ebx, ecx, edx) = self.helper.cpuid(eax, ecx)
|
|
logger().log_hal(f'[cpu] CPUID out: EAX=0x{eax:08X}, EBX=0x{ebx:08X}, ECX=0x{ecx:08X}, EDX=0x{edx:08X}')
|
|
return (eax, ebx, ecx, edx)
|
|
|
|
# Using cpuid check if running under vmm control
|
|
def check_vmm(self) -> int:
|
|
# check Hypervisor Present
|
|
(_, ebx, ecx, edx) = self.cpuid(0x01, 0)
|
|
if (ecx & 0x80000000):
|
|
(_, ebx, ecx, edx) = self.cpuid(0x40000000, 0)
|
|
is_xen = ((ebx == 0x566e6558) and (ecx == 0x65584d4d) and (edx == 0x4d4d566e))
|
|
if is_xen:
|
|
return VMM_XEN
|
|
is_hyperv = ((ebx == 0x7263694D) and (ecx == 0x666F736F) and (edx == 0x76482074))
|
|
if is_hyperv:
|
|
return VMM_HYPER_V
|
|
is_vmware = ((ebx == 0x61774d56) and (ecx == 0x4d566572) and (edx == 0x65726177))
|
|
if is_vmware:
|
|
return VMM_VMWARE
|
|
is_kvm = ((ebx == 0x4b4d564b) and (ecx == 0x564b4d56) and (edx == 0x0000004d))
|
|
if is_kvm:
|
|
return VMM_KVM
|
|
return VMM_NONE
|
|
|
|
# Using CPUID we can determine if Hyper-Threading is enabled in the CPU
|
|
def is_HT_active(self) -> bool:
|
|
logical_processor_per_core = self.get_number_logical_processor_per_core()
|
|
return logical_processor_per_core > 1
|
|
|
|
# Using the CPUID we determine the number of logical processors per core
|
|
def get_number_logical_processor_per_core(self) -> int:
|
|
(_, ebx, _, _) = self.cpuid(0x0b, 0x0)
|
|
return ebx
|
|
|
|
# Using CPUID we can determine the number of logical processors per package
|
|
def get_number_logical_processor_per_package(self) -> int:
|
|
(_, ebx, _, _) = self.cpuid(0x0b, 0x1)
|
|
return ebx
|
|
|
|
# Using CPUID we can determine the number of physical processors per package
|
|
def get_number_physical_processor_per_package(self) -> int:
|
|
logical_processor_per_core = self.get_number_logical_processor_per_core()
|
|
logical_processor_per_package = self.get_number_logical_processor_per_package()
|
|
return (logical_processor_per_package // logical_processor_per_core)
|
|
|
|
# determine number of logical processors in the core
|
|
def get_number_threads_from_APIC_table(self) -> int:
|
|
_acpi = self.cs.hals.acpi
|
|
dACPIID = {}
|
|
for apic in _acpi.get_parse_ACPI_table(acpi.ACPI_TABLE_SIG_APIC): # (table_header, APIC_object, table_header_blob, table_blob)
|
|
_, APIC_object, _, _ = apic
|
|
for structure in APIC_object.apic_structs:
|
|
if 0x00 == structure.Type:
|
|
if not structure.ACICID in dACPIID:
|
|
if 1 == structure.Flags:
|
|
dACPIID[structure.APICID] = structure.ACPIProcID
|
|
return len(dACPIID)
|
|
|
|
# determine the cpu threads location within a package/core
|
|
def get_cpu_topology(self) -> Dict[str, Dict[int, List[int]]]:
|
|
num_threads = self.cs.helper.get_threads_count()
|
|
packages: Dict[int, List[int]] = {}
|
|
cores: Dict[int, List[int]] = {}
|
|
for thread in range(num_threads):
|
|
if num_threads > 1:
|
|
self.logger.log_hal(f'Setting affinity to: {thread:d}')
|
|
self.cs.helper.set_affinity(thread)
|
|
eax = 0xb # cpuid leaf 0B contains x2apic info
|
|
ecx = 1 # ecx 1 will get us pkg_id in edx after shifting right by _eax
|
|
(_eax, _, _, _edx) = self.cpuid(eax, ecx)
|
|
pkg_id = _edx >> (_eax & 0xf)
|
|
if pkg_id not in packages:
|
|
packages[pkg_id] = []
|
|
packages[pkg_id].append(thread)
|
|
|
|
ecx = 0 # ecx 0 will get us the core_id in edx after shifting right by _eax
|
|
(_eax, _, _, _edx) = self.cpuid(eax, ecx)
|
|
core_id = _edx >> (_eax & 0xf)
|
|
if core_id not in cores:
|
|
cores[core_id] = []
|
|
cores[core_id].append(thread)
|
|
self.logger.log_hal(f'pkg id is {pkg_id:x}')
|
|
self.logger.log_hal(f'core id is {core_id:x}')
|
|
topology = {'packages': packages, 'cores': cores, 'threads':num_threads}
|
|
return topology
|
|
|
|
# determine number of physical sockets using the CPUID and APIC ACPI table
|
|
def get_number_sockets_from_APIC_table(self) -> int:
|
|
number_threads = self.get_number_threads_from_APIC_table()
|
|
logical_processor_per_package = self.get_number_logical_processor_per_package()
|
|
return (number_threads // logical_processor_per_package)
|
|
|
|
#
|
|
# Return SMRR MSR physical base and mask
|
|
#
|
|
def get_SMRR(self) -> Tuple[int, int]:
|
|
smrambase = self.cs.register.get_list_by_name('8086.MSR.IA32_SMRR_PHYSBASE').read_field('PHYSBASE', True)[0]
|
|
smrrmask = self.cs.register.get_list_by_name('8086.MSR.IA32_SMRR_PHYSMASK').read_field('PHYSMASK', True)[0]
|
|
|
|
return (smrambase, smrrmask)
|
|
|
|
#
|
|
# Return SMRAM region base, limit and size as defined by SMRR
|
|
#
|
|
def get_SMRR_SMRAM(self) -> Tuple[int, int, int]:
|
|
(smram_base, smrrmask) = self.get_SMRR()
|
|
smram_base &= smrrmask
|
|
smram_size = ((~smrrmask) & 0xFFFFFFFF) + 1
|
|
smram_limit = smram_base + smram_size - 1
|
|
return (smram_base, smram_limit, smram_size)
|
|
|
|
#
|
|
# Returns TSEG base, limit and size
|
|
#
|
|
def get_TSEG(self) -> Tuple[int, int, int]:
|
|
if self.cs.is_server():
|
|
# tseg register has base and limit
|
|
tseg_base = self.cs.register.get_list_by_name('8086.MEMMAP_VTD.TSEG_BASE').read_field('base', True)[0]
|
|
tseg_limit = self.cs.register.get_list_by_name('8086.MEMMAP_VTD.TSEG_LIMIT').read_field('limit', True)[0]
|
|
tseg_limit += 0xFFFFF
|
|
else:
|
|
# TSEG base is in TSEGMB, TSEG limit is BGSM - 1
|
|
tseg_base = self.cs.register.get_list_by_name('8086.HOSTCTL.TSEGMB').read_field('TSEGMB', True)[0]
|
|
bgsm = self.cs.register.get_list_by_name('8086.HOSTCTL.BGSM').read_field('BGSM', True)[0]
|
|
tseg_limit = bgsm - 1
|
|
|
|
tseg_size = tseg_limit - tseg_base + 1
|
|
|
|
return (tseg_base, tseg_limit, tseg_size)
|
|
|
|
#
|
|
# Returns SMRAM base from either SMRR MSR or TSEG PCIe config register
|
|
#
|
|
def get_SMRAM(self) -> Tuple[int, int, int]:
|
|
smram_base = None
|
|
smram_limit = None
|
|
smram_size = 0
|
|
try:
|
|
if self.check_SMRR_supported():
|
|
(smram_base, smram_limit, smram_size) = self.get_SMRR_SMRAM()
|
|
except Exception:
|
|
self.logger.log_hal('[cpu] Error using get_SMRR_SMRAM() to get SMRAM!')
|
|
|
|
if (smram_base is None) or (smram_limit is None):
|
|
try:
|
|
(smram_base, smram_limit, smram_size) = self.get_TSEG()
|
|
except Exception:
|
|
self.logger.log_hal('[cpu] Error using get_TSEG() to get SMRAM!')
|
|
smram_base = 0
|
|
smram_limit = 0
|
|
return (smram_base, smram_limit, smram_size)
|
|
|
|
#
|
|
# Check that SMRR is supported by CPU in IA32_MTRRCAP_MSR[SMRR]
|
|
#
|
|
def check_SMRR_supported(self) -> bool:
|
|
mtrrcap_msr_reg = self.cs.register.get_list_by_name('8086.MSR.MTRRCAP')
|
|
mtrrcap_msr_reg.read_and_hal_print()
|
|
return mtrrcap_msr_reg.is_any_field_value(1, 'SMRR')
|
|
|
|
#
|
|
# Dump CPU page tables at specified physical base of paging-directory hierarchy (CR3)
|
|
#
|
|
def dump_page_tables(self, cr3: int, pt_fname: Optional[str] = None) -> None:
|
|
_orig_logname = logger().LOG_FILE_NAME
|
|
hpt = paging.c_ia32e_page_tables(self.cs)
|
|
logger().log_hal(f'[cpu] dumping paging hierarchy at physical base (CR3) = 0x{cr3:08X}...')
|
|
if pt_fname is None:
|
|
pt_fname = f'pt_{cr3:08X}'
|
|
logger().set_log_file(pt_fname)
|
|
hpt.read_pt_and_show_status(pt_fname, 'PT', cr3)
|
|
logger().set_log_file(_orig_logname)
|
|
if hpt.failure:
|
|
logger().log_error('could not dump page tables')
|
|
|
|
def dump_page_tables_all(self) -> None:
|
|
for tid in range(self.cs.hals.msr.get_cpu_thread_count()):
|
|
cr3 = self.read_cr(tid, 3)
|
|
logger().log_hal(f'[cpu{tid:d}] found paging hierarchy base (CR3): 0x{cr3:08X}')
|
|
self.dump_page_tables(cr3)
|
|
|
|
|
|
haldata = {"arch":[hal_base.HALBase.MfgIds.Any], 'name': {'cpu': "CPU"}}
|