mirror of
https://gitlab.com/libvirt/libvirt.git
synced 2024-10-02 04:15:46 +00:00
0b9e70b141
When doing direct kernel boot we need to include the kernel, initrd and cmdline in the measurement. Reviewed-by: Cole Robinson <crobinso@redhat.com> Signed-off-by: Daniel P. Berrangé <berrange@redhat.com>
556 lines
18 KiB
Python
Executable File
556 lines
18 KiB
Python
Executable File
#!/usr/bin/python3
|
|
#
|
|
# SPDX-License-Identifier: LGPL-2.1-or-later
|
|
#
|
|
# Validates a guest AMD SEV launch measurement
|
|
#
|
|
# A general principle in writing this tool is that it must calculate the
|
|
# expected measurement based entirely on information it receives on the CLI
|
|
# from the guest owner.
|
|
#
|
|
# It cannot generally trust information obtained from the guest XML or from the
|
|
# virtualization host OS. The main exceptions are:
|
|
#
|
|
# - The guest measurement
|
|
#
|
|
# This is a result of cryptographic operation using a shared secret known
|
|
# only to the guest owner and SEV platform, not the host OS.
|
|
#
|
|
# - The guest policy
|
|
#
|
|
# This is encoded in the launch session blob that is encrypted with a shared
|
|
# secret known only to the guest owner and SEV platform, not the host OS. It
|
|
# is impossible for the host OS to maliciously launch a guest with different
|
|
# policy and the user provided launch session blob.
|
|
#
|
|
# CAVEAT: the user must ALWAYS create a launch blob with freshly generated
|
|
# TIK/TEK for every new VM. Re-use of the same TIK/TEK for multiple VMs
|
|
# is insecure.
|
|
#
|
|
# - The SEV API version / build ID
|
|
#
|
|
# This does not have an impact on the security of the measurement, unless
|
|
# the guest owner needs a guarantee that the host is not using specific
|
|
# firmware versions with known flaws.
|
|
#
|
|
|
|
import abc
|
|
import argparse
|
|
from base64 import b64decode
|
|
from hashlib import sha256
|
|
import hmac
|
|
import logging
|
|
import re
|
|
import socket
|
|
import sys
|
|
import traceback
|
|
from uuid import UUID
|
|
|
|
from lxml import etree
|
|
import libvirt
|
|
|
|
log = logging.getLogger()
|
|
|
|
|
|
class AttestationFailedException(Exception):
|
|
pass
|
|
|
|
|
|
class UnsupportedUsageException(Exception):
|
|
pass
|
|
|
|
|
|
class InsecureUsageException(Exception):
|
|
pass
|
|
|
|
|
|
class IncorrectConfigException(Exception):
|
|
pass
|
|
|
|
|
|
class InvalidStateException(Exception):
|
|
pass
|
|
|
|
|
|
class GUIDTable(abc.ABC):
|
|
GUID_LEN = 16
|
|
|
|
def __init__(self, guid, lenlen=2):
|
|
self.guid = guid
|
|
self.lenlen = lenlen
|
|
|
|
@abc.abstractmethod
|
|
def entries(self):
|
|
pass
|
|
|
|
def build_entry(self, guid, payload, lenlen):
|
|
dummylen = int(0).to_bytes(lenlen, 'little')
|
|
entry = bytearray(guid + dummylen + payload)
|
|
|
|
lenle = len(entry).to_bytes(lenlen, 'little')
|
|
entry[self.GUID_LEN:(self.GUID_LEN + lenlen)] = lenle
|
|
|
|
return bytes(entry)
|
|
|
|
def build(self):
|
|
payload = self.entries()
|
|
|
|
if len(payload) == 0:
|
|
return bytes([])
|
|
|
|
dummylen = int(0).to_bytes(self.lenlen, 'little')
|
|
table = bytearray(self.guid + dummylen + payload)
|
|
|
|
guidlen = len(table).to_bytes(self.lenlen, 'little')
|
|
table[self.GUID_LEN:(self.GUID_LEN + self.lenlen)] = guidlen
|
|
|
|
pad = 16 - (len(table) % 16)
|
|
table += bytes([0]) * pad
|
|
|
|
log.debug("Table(hex): %s", bytes(table).hex())
|
|
return bytes(table)
|
|
|
|
|
|
class KernelTable(GUIDTable):
|
|
|
|
TABLE_GUID = UUID('{9438d606-4f22-4cc9-b479-a793-d411fd21}').bytes_le
|
|
KERNEL_GUID = UUID('{4de79437-abd2-427f-b835-d5b1-72d2045b}').bytes_le
|
|
INITRD_GUID = UUID('{44baf731-3a2f-4bd7-9af1-41e2-9169781d}').bytes_le
|
|
CMDLINE_GUID = UUID('{97d02dd8-bd20-4c94-aa78-e771-4d36ab2a}').bytes_le
|
|
|
|
def __init__(self):
|
|
super().__init__(guid=self.TABLE_GUID,
|
|
lenlen=2)
|
|
|
|
self.kernel = None
|
|
self.initrd = None
|
|
self.cmdline = None
|
|
|
|
def load_kernel(self, path):
|
|
with open(path, "rb") as fh:
|
|
self.kernel = sha256(fh.read()).digest()
|
|
|
|
def load_initrd(self, path):
|
|
with open(path, "rb") as fh:
|
|
self.initrd = sha256(fh.read()).digest()
|
|
|
|
def load_cmdline(self, val):
|
|
self.cmdline = sha256(val.encode("utf8") + bytes([0])).digest()
|
|
|
|
def entries(self):
|
|
entries = bytes([])
|
|
if self.kernel is None:
|
|
return entries
|
|
|
|
if self.initrd is None:
|
|
self.initrd = sha256(bytes([])).digest()
|
|
if self.cmdline is None:
|
|
self.cmdline = sha256(bytes([0])).digest()
|
|
|
|
log.debug("Kernel(sha256): %s", self.kernel.hex())
|
|
log.debug("Initrd(sha256): %s", self.initrd.hex())
|
|
log.debug("Cmdline(sha256): %s", self.cmdline.hex())
|
|
entries += self.build_entry(self.CMDLINE_GUID, self.cmdline, 2)
|
|
entries += self.build_entry(self.INITRD_GUID, self.initrd, 2)
|
|
entries += self.build_entry(self.KERNEL_GUID, self.kernel, 2)
|
|
|
|
return entries
|
|
|
|
|
|
class ConfidentialVM(object):
|
|
|
|
def __init__(self,
|
|
measurement=None,
|
|
api_major=None,
|
|
api_minor=None,
|
|
build_id=None,
|
|
policy=None):
|
|
self.measurement = measurement
|
|
self.api_major = api_major
|
|
self.api_minor = api_minor
|
|
self.build_id = build_id
|
|
self.policy = policy
|
|
|
|
self.firmware = None
|
|
self.tik = None
|
|
self.tek = None
|
|
|
|
self.kernel_table = KernelTable()
|
|
|
|
def load_tik_tek(self, tik_path, tek_path):
|
|
with open(tik_path, 'rb') as fh:
|
|
self.tik = fh.read()
|
|
log.debug("TIK(hex): %s", self.tik.hex())
|
|
|
|
if len(self.tik) != 16:
|
|
raise UnsupportedUsageException(
|
|
"Expected 16 bytes in TIK file, but got %d" % len(self.tik))
|
|
|
|
with open(tek_path, 'rb') as fh:
|
|
self.tek = fh.read()
|
|
log.debug("TEK(hex): %s", self.tek.hex())
|
|
|
|
if len(self.tek) != 16:
|
|
raise UnsupportedUsageException(
|
|
"Expected 16 bytes in TEK file, but got %d" % len(self.tek))
|
|
|
|
def load_tk(self, tk_path):
|
|
with open(tk_path, 'rb') as fh:
|
|
tk = fh.read()
|
|
|
|
if len(tk) != 32:
|
|
raise UnsupportedUsageException(
|
|
"Expected 32 bytes in TIK/TEK file, but got %d" % len(tk))
|
|
|
|
self.tek = tk[0:16]
|
|
self.tik = tk[16:32]
|
|
log.debug("TIK(hex): %s", self.tik.hex())
|
|
log.debug("TEK(hex): %s", self.tek.hex())
|
|
|
|
def load_firmware(self, firmware_path):
|
|
with open(firmware_path, 'rb') as fh:
|
|
self.firmware = fh.read()
|
|
log.debug("Firmware(sha256): %s", sha256(self.firmware).hexdigest())
|
|
|
|
# Get the full set of measured launch data for the domain
|
|
#
|
|
# The measured data that the guest is initialized with is the concatenation
|
|
# of the following:
|
|
#
|
|
# - The firmware blob
|
|
# - The kernel GUID table
|
|
def get_measured_data(self):
|
|
measured_data = (self.firmware +
|
|
self.kernel_table.build())
|
|
log.debug("Measured-data(sha256): %s",
|
|
sha256(measured_data).hexdigest())
|
|
return measured_data
|
|
|
|
# Get the reported and computed launch measurements for the domain
|
|
#
|
|
# AMD Secure Encrypted Virtualization API , section 6.5:
|
|
#
|
|
# measurement = HMAC(0x04 || API_MAJOR || API_MINOR || BUILD ||
|
|
# GCTX.POLICY || GCTX.LD || MNONCE; GCTX.TIK)
|
|
#
|
|
# Where GCTX.LD covers all the measured data the guest is initialized with
|
|
# per get_measured_data().
|
|
def get_measurements(self):
|
|
measurement = b64decode(self.measurement)
|
|
reported = measurement[0:32]
|
|
nonce = measurement[32:48]
|
|
|
|
measured_data = self.get_measured_data()
|
|
msg = (
|
|
bytes([0x4]) +
|
|
self.api_major.to_bytes(1, 'little') +
|
|
self.api_minor.to_bytes(1, 'little') +
|
|
self.build_id.to_bytes(1, 'little') +
|
|
self.policy.to_bytes(4, 'little') +
|
|
sha256(measured_data).digest() +
|
|
nonce
|
|
)
|
|
log.debug("Measured-msg(hex): %s", msg.hex())
|
|
|
|
computed = hmac.new(self.tik, msg, 'sha256').digest()
|
|
|
|
log.debug("Measurement reported(hex): %s", reported.hex())
|
|
log.debug("Measurement computed(hex): %s", computed.hex())
|
|
|
|
return reported, computed
|
|
|
|
def attest(self):
|
|
reported, computed = self.get_measurements()
|
|
|
|
if reported != computed:
|
|
raise AttestationFailedException(
|
|
"Measurement does not match, VM is not trustworthy")
|
|
|
|
|
|
class LibvirtConfidentialVM(ConfidentialVM):
|
|
def __init__(self, **kwargs):
|
|
super().__init__(**kwargs)
|
|
|
|
self.conn = None
|
|
self.dom = None
|
|
|
|
def check_domain(self, doc, secure):
|
|
ls = doc.xpath("/domain/launchSecurity[@type='sev']")
|
|
if len(ls) != 1:
|
|
raise IncorrectConfigException(
|
|
"Domain is not configured with SEV launch security")
|
|
|
|
dh = doc.xpath("/domain/launchSecurity[@type='sev']/dhCert")
|
|
if len(dh) != 1:
|
|
raise IncorrectConfigException(
|
|
"Domain must have SEV owner cert to validate measurement")
|
|
|
|
session = doc.xpath("/domain/launchSecurity[@type='sev']/session")
|
|
if len(session) != 1:
|
|
raise IncorrectConfigException(
|
|
"Domain must have SEV session data to validate measurement")
|
|
|
|
nvramnodes = doc.xpath("/domain/os/nvram")
|
|
if len(nvramnodes) != 0 and secure:
|
|
raise InsecureUsageException(
|
|
"Domain firmware with NVRAM cannot be securely measured")
|
|
|
|
loadernodes = doc.xpath("/domain/os/loader")
|
|
if len(loadernodes) != 1:
|
|
raise IncorrectConfigException(
|
|
"Domain must have one firmware path")
|
|
|
|
def load_domain(self, uri, id_name_uuid, secure, ignore_config):
|
|
self.conn = libvirt.open(uri)
|
|
|
|
remote = socket.gethostname() != self.conn.getHostname()
|
|
if not remote and secure:
|
|
raise InsecureUsageException(
|
|
"running locally on the hypervisor host is not secure")
|
|
|
|
if re.match(r'^\d+$', id_name_uuid):
|
|
self.dom = self.conn.lookupByID(int(id_name_uuid))
|
|
elif re.match(r'^[-a-f0-9]+$', id_name_uuid):
|
|
self.dom = self.conn.lookupByUUIDString(id_name_uuid)
|
|
else:
|
|
self.dom = self.conn.lookupByName(id_name_uuid)
|
|
|
|
log.debug("VM: id=%d name=%s uuid=%s",
|
|
self.dom.ID(), self.dom.name(), self.dom.UUIDString())
|
|
|
|
if not self.dom.isActive():
|
|
raise InvalidStateException(
|
|
"Domain must be running to validate measurement")
|
|
|
|
state = self.dom.info()[0]
|
|
if state != libvirt.VIR_DOMAIN_PAUSED and secure:
|
|
raise InvalidStateException(
|
|
"Domain must be paused to validate measurement")
|
|
|
|
xml = self.dom.XMLDesc()
|
|
|
|
doc = etree.fromstring(xml)
|
|
if not ignore_config:
|
|
self.check_domain(doc, secure)
|
|
|
|
# See comments at top of file wrt why we are OK to trust the
|
|
# sev-api-major, sev-api-minor, sev-build-id and sev-policy data
|
|
# reported by the host
|
|
sevinfo = self.dom.launchSecurityInfo()
|
|
|
|
if "sev-api-major" not in sevinfo:
|
|
raise UnsupportedUsageException(
|
|
"'api-major' not reported in domain launch security info")
|
|
|
|
if self.measurement is None:
|
|
self.measurement = sevinfo["sev-measurement"]
|
|
if self.api_major is None:
|
|
self.api_major = sevinfo["sev-api-major"]
|
|
if self.api_minor is None:
|
|
self.api_minor = sevinfo["sev-api-minor"]
|
|
if self.build_id is None:
|
|
self.build_id = sevinfo["sev-build-id"]
|
|
if self.policy is None:
|
|
self.policy = sevinfo["sev-policy"]
|
|
|
|
if self.firmware is None:
|
|
if remote:
|
|
raise UnsupportedUsageException(
|
|
"Cannot access firmware path remotely")
|
|
if secure:
|
|
raise InsecureUsageException(
|
|
"Using firmware path from XML is not secure")
|
|
|
|
loadernodes = doc.xpath("/domain/os/loader")
|
|
if len(loadernodes) == 0:
|
|
raise UnsupportedUsageException(
|
|
"--firmware not specified and no firmware path found")
|
|
|
|
self.load_firmware(loadernodes[0].text)
|
|
|
|
|
|
def parse_command_line():
|
|
parser = argparse.ArgumentParser(
|
|
description='Validate guest AMD SEV launch measurement')
|
|
parser.add_argument('--debug', '-d', action='store_true',
|
|
help='Show debug information')
|
|
parser.add_argument('--quiet', '-q', action='store_true',
|
|
help='Do not display status')
|
|
|
|
# Arguments related to the state of the launched guest
|
|
vmstate = parser.add_argument_group("Virtual machine launch state")
|
|
vmstate.add_argument('--measurement', '-m',
|
|
help='Measurement for the running domain')
|
|
vmstate.add_argument('--api-major', type=int,
|
|
help='SEV API major version for the running domain')
|
|
vmstate.add_argument('--api-minor', type=int,
|
|
help='SEV API minor version for the running domain')
|
|
vmstate.add_argument('--build-id', type=int,
|
|
help='SEV build ID for the running domain')
|
|
vmstate.add_argument('--policy', type=int,
|
|
help='SEV policy for the running domain')
|
|
|
|
# Arguments related to calculation of the expected launch measurement
|
|
vmconfig = parser.add_argument_group("Virtual machine config")
|
|
vmconfig.add_argument('--firmware', '-f',
|
|
help='Path to the firmware binary')
|
|
vmconfig.add_argument('--kernel', '-k',
|
|
help='Path to the kernel binary')
|
|
vmconfig.add_argument('--initrd', '-r',
|
|
help='Path to the initrd binary')
|
|
vmconfig.add_argument('--cmdline', '-e',
|
|
help='Cmdline string booted with')
|
|
vmconfig.add_argument('--tik',
|
|
help='TIK file for domain')
|
|
vmconfig.add_argument('--tek',
|
|
help='TEK file for domain')
|
|
vmconfig.add_argument('--tk',
|
|
help='TEK/TIK combined file for domain')
|
|
|
|
# Arguments related to the connection to libvirt
|
|
vmconn = parser.add_argument_group("Libvirt guest connection")
|
|
vmconn.add_argument('--connect', '-c', default="qemu:///system",
|
|
help='libvirt connection URI')
|
|
vmconn.add_argument('--domain', '-o',
|
|
help='domain ID / Name / UUID')
|
|
vmconn.add_argument('--insecure', '-i', action='store_true',
|
|
help='Proceed even if usage scenario is insecure')
|
|
vmconn.add_argument('--ignore-config', '-g', action='store_true',
|
|
help='Do not attempt to sanity check the guest config')
|
|
|
|
return parser.parse_args()
|
|
|
|
|
|
# Sanity check the set of CLI args specified provide enough info for us to do
|
|
# the job
|
|
def check_usage(args):
|
|
if args.tk is not None:
|
|
if args.tik is not None or args.tek is not None:
|
|
raise UnsupportedUsageException(
|
|
"--tk is mutually exclusive with --tek/--tik")
|
|
else:
|
|
if args.tik is None or args.tek is None:
|
|
raise UnsupportedUsageException(
|
|
"Either --tk or both of --tek/--tik are required")
|
|
|
|
if args.domain is None:
|
|
if args.measurement is None:
|
|
raise UnsupportedUsageException(
|
|
"Either --measurement or --domain is required")
|
|
|
|
if args.api_major is None:
|
|
raise UnsupportedUsageException(
|
|
"Either --api-major or --domain is required")
|
|
|
|
if args.api_minor is None:
|
|
raise UnsupportedUsageException(
|
|
"Either --api-minor or --domain is required")
|
|
|
|
if args.build_id is None:
|
|
raise UnsupportedUsageException(
|
|
"Either --build-id or --domain is required")
|
|
|
|
if args.policy is None:
|
|
raise UnsupportedUsageException(
|
|
"Either --policy or --domain is required")
|
|
|
|
if args.firmware is None:
|
|
raise UnsupportedUsageException(
|
|
"Either --firmware or --domain is required")
|
|
|
|
if args.kernel is None:
|
|
if args.initrd is not None or args.cmdline is not None:
|
|
raise UnsupportedUsageException(
|
|
"--initrd/--cmdline require --kernel")
|
|
|
|
|
|
def attest(args):
|
|
if args.domain is None:
|
|
cvm = ConfidentialVM(measurement=args.measurement,
|
|
api_major=args.api_major,
|
|
api_minor=args.api_minor,
|
|
build_id=args.build_id,
|
|
policy=args.policy)
|
|
else:
|
|
cvm = LibvirtConfidentialVM(measurement=args.measurement,
|
|
api_major=args.api_major,
|
|
api_minor=args.api_minor,
|
|
build_id=args.build_id,
|
|
policy=args.policy)
|
|
|
|
if args.firmware is not None:
|
|
cvm.load_firmware(args.firmware)
|
|
|
|
if args.tk is not None:
|
|
cvm.load_tk(args.tk)
|
|
else:
|
|
cvm.load_tik_tek(args.tik, args.tek)
|
|
|
|
if args.kernel is not None:
|
|
cvm.kernel_table.load_kernel(args.kernel)
|
|
|
|
if args.initrd is not None:
|
|
cvm.kernel_table.load_initrd(args.initrd)
|
|
|
|
if args.cmdline is not None:
|
|
cvm.kernel_table.load_cmdline(args.cmdline)
|
|
|
|
if args.domain is not None:
|
|
cvm.load_domain(args.connect,
|
|
args.domain,
|
|
not args.insecure,
|
|
args.ignore_config)
|
|
|
|
cvm.attest()
|
|
|
|
if not args.quiet:
|
|
print("OK: Looks good to me")
|
|
|
|
def main():
|
|
args = parse_command_line()
|
|
if args.debug:
|
|
logging.basicConfig(level="DEBUG")
|
|
formatter = logging.Formatter("[%(levelname)s]: %(message)s")
|
|
handler = log.handlers[0]
|
|
handler.setFormatter(formatter)
|
|
|
|
try:
|
|
check_usage(args)
|
|
|
|
attest(args)
|
|
|
|
sys.exit(0)
|
|
except AttestationFailedException as e:
|
|
if args.debug:
|
|
traceback.print_tb(e.__traceback__)
|
|
if not args.quiet:
|
|
print("ERROR: %s" % e, file=sys.stderr)
|
|
sys.exit(1)
|
|
except UnsupportedUsageException as e:
|
|
if args.debug:
|
|
traceback.print_tb(e.__traceback__)
|
|
if not args.quiet:
|
|
print("ERROR: %s" % e, file=sys.stderr)
|
|
sys.exit(2)
|
|
except InsecureUsageException as e:
|
|
if not args.quiet:
|
|
print("ERROR: %s" % e, file=sys.stderr)
|
|
sys.exit(3)
|
|
except IncorrectConfigException as e:
|
|
if not args.quiet:
|
|
print("ERROR: %s" % e, file=sys.stderr)
|
|
sys.exit(4)
|
|
except InvalidStateException as e:
|
|
if not args.quiet:
|
|
print("ERROR: %s" % e, file=sys.stderr)
|
|
sys.exit(5)
|
|
except Exception as e:
|
|
if args.debug:
|
|
traceback.print_tb(e.__traceback__)
|
|
if not args.quiet:
|
|
print("ERROR: %s" % e, file=sys.stderr)
|
|
sys.exit(6)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|