diff --git a/tests/lib/Qemu.py b/tests/lib/Qemu.py index d1a52ac..d151f87 100644 --- a/tests/lib/Qemu.py +++ b/tests/lib/Qemu.py @@ -60,7 +60,7 @@ def __init__(self): def args(self): smp = ['-smp', f'{self.nb_cores},sockets={self.nb_sockets}'] if self.cpu_flags != '': - cpu = ['-cpu', self.cpu_type, self.cpu_flags] + cpu = ['-cpu', self.cpu_type + self.cpu_flags] else: cpu = ['-cpu', self.cpu_type] return cpu + smp @@ -478,6 +478,13 @@ def _setup_workdir(self): self.workdir = tempfile.TemporaryDirectory(prefix=f'tdxtest-{self.name}-') self.workdir_name = self.workdir.name + @property + def pid(self): + cs = subprocess.run(['cat', f'{self.workdir.name}/qemu.pid'], capture_output=True) + assert cs.returncode == 0, 'Failed getting qemu pid' + pid = int(cs.stdout.strip()) + return pid + def rsync_file(self, fname, dest, sudo=False): """ fname : local file or folder @@ -506,6 +513,17 @@ def run(self): stdout=subprocess.PIPE, stderr=subprocess.PIPE) + def run_and_wait(self): + """ + Run qemu and wait for its start (by waiting for monitor file's availability) + """ + cmd = self.qcmd.get_command() + print(' '.join(cmd)) + self.proc = subprocess.Popen(cmd, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE) + QemuMonitor(self) + def communicate(self): """ Wait for qemu to exit diff --git a/tests/tests/setup-env-tox.sh b/tests/tests/setup-env-tox.sh index 5e60853..f9db044 100755 --- a/tests/tests/setup-env-tox.sh +++ b/tests/tests/setup-env-tox.sh @@ -38,7 +38,12 @@ cleanup() { install_deps() { sudo apt install -y python3-parameterized \ sshpass \ - python3-cpuinfo &> /dev/null + cpuid \ + python3-cpuinfo + # Make sure kvm_intel is installed properly + # (one of the tests installed it with NOEPT) + sudo rmmod kvm_intel || true + sudo modprobe kvm_intel } rm -rf /var/tmp/tdxtest @@ -59,7 +64,8 @@ else fi fi - cleanup -install_deps &> /dev/null +set -e + +install_deps diff --git a/tests/tests/test_guest_cpu_off.py b/tests/tests/test_guest_cpu_off.py new file mode 100644 index 0000000..9748f55 --- /dev/null +++ b/tests/tests/test_guest_cpu_off.py @@ -0,0 +1,106 @@ +#!/usr/bin/env python3 +# +# Copyright 2024 Canonical Ltd. +# Authors: +# - Hector Cao +# +# This program is free software: you can redistribute it and/or modify it +# under the terms of the GNU General Public License version 3, as published +# by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranties of MERCHANTABILITY, +# SATISFACTORY QUALITY, 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, see . +# + +import os +import subprocess +import time +import re +import multiprocessing +import random + +import Qemu +import util + +script_path=os.path.dirname(os.path.realpath(__file__)) + +def test_guest_cpu_off(): + """ + tdx_VMP_cpu_onoff test case (See https://github.com/intel/tdx/wiki/Tests) + """ + + # Startup Qemu and connect via SSH + qm = Qemu.QemuMachine() + qm.run() + m = Qemu.QemuSSH(qm) + + # turn off arbitrary cpus + cpu = cpu_off_random() + + # make sure the VM still does things + still_working = True + try: + m.check_exec('ls /tmp') + except Exception as e: + still_working = False + + qm.stop() + + # turn back on cpus + cpu_on_off(f'/sys/devices/system/cpu/cpu{cpu}/online', 1) + + assert still_working, 'VM dysfunction when a cpu is brought offline' + +def test_guest_cpu_pinned_off(): + """ + tdx_cpuoff_pinedVMdown test case (See https://github.com/intel/tdx/wiki/Tests) + """ + + # do 20 iterations of starting up a VM, pinning the VM pid, turning off + # the pinned cpu and making sure host still works + for i in range(1,20): + print(f'Iteration: {i}') + qm = Qemu.QemuMachine() + qm.run_and_wait() + + cpu = pin_process_on_random_cpu(qm.pid) + + cpu_on_off(f'/sys/devices/system/cpu/cpu{cpu}/online', 0) + + m = Qemu.QemuSSH(qm) + m.check_exec('sudo init 0 &') + + qm.stop() + + cpu_on_off(f'/sys/devices/system/cpu/cpu{cpu}/online', 1) + +def pin_process_on_random_cpu(pid): + # pin pid to a particular cpu + cpu = cpu_select() + cs = subprocess.run(['sudo', 'taskset', '-pc', f'{cpu}', f'{pid}'], capture_output=True) + assert cs.returncode == 0, 'Failed pinning qemu pid to cpu 18' + return cpu + +def cpu_off_random(): + cpu = cpu_select() + cpu_on_off(f'/sys/devices/system/cpu/cpu{cpu}/online', 1) + cpu_on_off(f'/sys/devices/system/cpu/cpu{cpu}/online', 0) + return cpu + +def cpu_select(): + cpu_count = multiprocessing.cpu_count() + cpu = random.randint(0, cpu_count-1) + return cpu + +# Helper function for turning cpu on/off +def cpu_on_off(file_str, val): + dev_f = open(file_str, 'w') + cs = subprocess.run(['echo', f'{val}'], check=True, stdout=dev_f) + assert cs.returncode == 0, 'Failed turning cpu off' + dev_f.close() + diff --git a/tests/tests/test_guest_fail.py b/tests/tests/test_guest_fail.py new file mode 100644 index 0000000..aa9ddc7 --- /dev/null +++ b/tests/tests/test_guest_fail.py @@ -0,0 +1,84 @@ +#!/usr/bin/env python3 +# +# Copyright 2024 Canonical Ltd. +# Authors: +# - Hector Cao +# +# This program is free software: you can redistribute it and/or modify it +# under the terms of the GNU General Public License version 3, as published +# by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranties of MERCHANTABILITY, +# SATISFACTORY QUALITY, 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, see . +# + +import os +import subprocess +import time +import re + +import Qemu +import util + +script_path=os.path.dirname(os.path.realpath(__file__)) + +def test_guest_noept_fail(): + """ + tdx_NOEPT test case (See https://github.com/intel/tdx/wiki/Tests) + """ + + # Get initial dmesg contents for comparison later + cs = subprocess.run(['sudo', 'dmesg'], check=True, capture_output=True) + assert cs.returncode == 0, 'Failed getting dmesg' + dmesg_start = str(cs.stdout) + + with KvmIntelModuleReloader('pt_mode=1 ept=0') as module: + # Get after modprobe dmesg contents + cs = subprocess.run(['sudo', 'dmesg'], check=True, capture_output=True) + assert cs.returncode == 0, 'Failed getting dmesg' + dmesg_end = str(cs.stdout) + + # Verify "TDX requires mmio caching" in dmesg (but only one more time) + dmesg_start_count = dmesg_start.count("TDX requires TDP MMU. Please enable TDP MMU for TDX") + dmesg_end_count = dmesg_end.count("TDX requires TDP MMU. Please enable TDP MMU for TDX") + assert dmesg_end_count == dmesg_start_count+1, "dmesg missing proper message" + + # Run Qemu and verify failure + qm = Qemu.QemuMachine() + qm.run() + + # expect qemu quit immediately with specific error message + _, err = qm.communicate() + assert "-accel kvm: vm-type tdx not supported by KVM" in err.decode() + +def test_guest_disable_tdx_fail(): + """ + tdx_disabled test case (See https://github.com/intel/tdx/wiki/Tests) + """ + + with KvmIntelModuleReloader('tdx=0') as module: + # Run Qemu and verify failure + qm = Qemu.QemuMachine() + qm.run() + + # expect qemu quit immediately with specific error message + _, err = qm.communicate() + assert "-accel kvm: vm-type tdx not supported by KVM" in err.decode() + +class KvmIntelModuleReloader: + """ + kvm_intel module reloader (context manager) + Allow to reload kvm_intel module with custom arguments + """ + def __init__(self, module_args=''): + self.args = module_args + def __enter__(self): + subprocess.check_call('sudo rmmod kvm_intel', shell=True) + subprocess.check_call(f'sudo modprobe kvm_intel {self.args}', shell=True) + def __exit__(self, exc_type, exc_value, exc_tb): + subprocess.check_call('sudo modprobe kvm_intel', shell=True) diff --git a/tests/tests/test_guest_tsc_config.py b/tests/tests/test_guest_tsc_config.py new file mode 100644 index 0000000..1300041 --- /dev/null +++ b/tests/tests/test_guest_tsc_config.py @@ -0,0 +1,134 @@ +#!/usr/bin/env python3 +# +# Copyright 2024 Canonical Ltd. +# Authors: +# - Hector Cao +# +# This program is free software: you can redistribute it and/or modify it +# under the terms of the GNU General Public License version 3, as published +# by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranties of MERCHANTABILITY, +# SATISFACTORY QUALITY, 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, see . +# + +import os +import subprocess +import time +import re + +import Qemu +import util + +script_path=os.path.dirname(os.path.realpath(__file__)) + +def test_guest_tsc_config(): + """ + tdx_tsc_config test case (See https://github.com/intel/tdx/wiki/Tests) + """ + + # Get and parse cpuid value from host + cs = subprocess.run(['cpuid', '-rl', '0x15', '-1'], check=True, capture_output=True) + assert cs.returncode == 0, 'Failed getting cpuid' + out_str = str(cs.stdout.strip()) + eax, ebx, ecx, edx = parse_cpuid_0x15_values(out_str) + assert edx == 0, "CPUID values incorrect" + + # calculate tsc value + tsc_host = ecx * ebx / eax + + qm = Qemu.QemuMachine() + qm.run() + + # Get cpuid value from guest and parse it + m = Qemu.QemuSSH(qm) + out_str = '' + [outlines, err] = m.check_exec('cpuid -rl 0x15 -1') + for l in outlines.readlines(): + out_str += l + eax, ebx, ecx, edx = parse_cpuid_0x15_values(out_str) + + # calculate tsc value on guest and make sure same as host + tsc_guest = ecx * ebx / eax + assert tsc_guest == tsc_host, "TSC host and guest don't match" + + # Verify tsc detected in dmesg logs + cs = subprocess.run(['sudo', 'dmesg'], check=True, capture_output=True) + assert cs.returncode == 0, 'Failed getting dmesg' + assert "tsc: Detected" in str(cs.stdout), "tsc not found in dmesg" + + qm.stop() + + +def test_guest_set_tsc_frequency(): + """ + tdx_tsc_config test case (See https://github.com/intel/tdx/wiki/Tests) + """ + + # Set guest tsc frequency + tsc_frequency = 3000000000 + qm = Qemu.QemuMachine() + qm.qcmd.plugins['cpu'].cpu_flags += f',tsc-freq={tsc_frequency}' + qm.run() + + # Get cpuid value from guest and parse it + m = Qemu.QemuSSH(qm) + out_str = '' + [outlines, err] = m.check_exec('cpuid -rl 0x15 -1') + for l in outlines.readlines(): + out_str += l + eax, ebx, ecx, edx = parse_cpuid_0x15_values(out_str) + + # calculate tsc on guest and make sure its equal to value set + tsc_guest = ecx * ebx / eax + assert tsc_guest == tsc_frequency, "TSC frequency not set correctly" + +def test_guest_tsc_deadline_enable(): + """ + tdx_tsc_deadline_enable test case (See https://github.com/intel/tdx/wiki/Tests) + """ + qm = Qemu.QemuMachine() + qm.run() + + m = Qemu.QemuSSH(qm) + + stdout, _ = m.check_exec('lscpu') + output = stdout.read().decode('utf-8') + assert 'Flags' in output + assert 'tsc_deadline_timer' in output + + qm.stop() + +def test_guest_tsc_deadline_disable(): + """ + tdx_tsc_deadline_disable test case (See https://github.com/intel/tdx/wiki/Tests) + """ + qm = Qemu.QemuMachine() + qm.qcmd.plugins['cpu'].cpu_flags += f',-tsc-deadline' + qm.run() + + m = Qemu.QemuSSH(qm) + + stdout, _ = m.check_exec('lscpu') + output = stdout.read().decode('utf-8') + assert 'Flags' in output + assert 'tsc_deadline_timer' not in output + + qm.stop() + +# helper function for parsing cpuid value into registers +def parse_cpuid_0x15_values(val_str): + # parse register values + try: + eax = int(re.findall(r'eax=0x([0-9a-f]+)', val_str)[0], 16) + ebx = int(re.findall(r'ebx=0x([0-9a-f]+)', val_str)[0], 16) + ecx = int(re.findall(r'ecx=0x([0-9a-f]+)', val_str)[0], 16) + edx = int(re.findall(r'edx=0x([0-9a-f]+)', val_str)[0], 16) + except Exception as e: + assert False, print(f'Failed parsing cpuid registers {e}') + return eax, ebx, ecx, edx diff --git a/tests/tests/test_host_tdx_software.py b/tests/tests/test_host_tdx_software.py index 992a21a..6da8333 100644 --- a/tests/tests/test_host_tdx_software.py +++ b/tests/tests/test_host_tdx_software.py @@ -17,6 +17,7 @@ # this program. If not, see . # +import re import subprocess def test_host_tdx_software(): @@ -28,5 +29,21 @@ def test_host_tdx_software(): subprocess.check_call('grep Y /sys/module/kvm_intel/parameters/sgx', shell=True) +def test_host_tdx_module_load(): + """ + Check the tdx module has been loaded successfuly on the host + Check a log in dmesg with appropriate versioning information + + tdx_uefi test case (See https://github.com/intel/tdx/wiki/Tests) + """ + + # Get dmesg and make sure it has the tdx module load message + cs = subprocess.run(['sudo', 'dmesg'], check=True, capture_output=True) + assert cs.returncode == 0, 'Failed getting dmesg' + dmesg_str = cs.stdout.decode('utf-8') + + items=re.findall(r'tdx: TDX module: attributes 0x[0-9]+, vendor_id 0x8086, major_version [0-9]+, minor_version [0-9]+, build_date [0-9]+, build_num [0-9]+', dmesg_str) + assert len(items) > 0 + if __name__ == '__main__': test_host_tdx_software() diff --git a/tests/tox.ini b/tests/tox.ini index c16542a..4a930c4 100644 --- a/tests/tox.ini +++ b/tests/tox.ini @@ -5,7 +5,9 @@ isolated_build = True skip_missing_interpreters = True [testenv] -allowlist_externals=bash +allowlist_externals = + bash + {toxinidir}/tests/setup-env-tox.sh pass_env = TDXTEST_GUEST_IMG, TDXTEST_DEBUG envdir = {toxworkdir}/.venv deps = @@ -18,7 +20,7 @@ setenv = PYTHONPATH={env:PYTHONPATH}:{toxinidir}/lib commands_pre = bash -c 'if [ `id -u` -ne 0 ]; then echo "Must run all tests with sudo" 1>&2; (exit -1); fi' - bash -c {toxinidir}/tests/setup-env-tox.sh + {toxinidir}/tests/setup-env-tox.sh [testenv:test_guest] commands =