blob: f2168c18a3dc729c7ed54bc1d21fe2ea9e74a185 [file] [log] [blame]
#!/usr/bin/env python3
# Handle running OE images standalone with QEMU
# Copyright (C) 2006-2011 Linux Foundation
# Copyright (c) 2016 Wind River Systems, Inc.
# SPDX-License-Identifier: GPL-2.0-only
import os
import sys
import logging
import subprocess
import re
import fcntl
import shutil
import glob
import configparser
import signal
class RunQemuError(Exception):
"""Custom exception to raise on known errors."""
class OEPathError(RunQemuError):
"""Custom Exception to give better guidance on missing binaries"""
def __init__(self, message):
super().__init__("In order for this script to dynamically infer paths\n \
kernels or filesystem images, you either need bitbake in your PATH\n \
or to source oe-init-build-env before running this script.\n\n \
Dynamic path inference can be avoided by passing a *.qemuboot.conf to\n \
runqemu, i.e. `runqemu /path/to/my-image-name.qemuboot.conf`\n\n %s" % message)
def create_logger():
logger = logging.getLogger('runqemu')
# create console handler and set level to debug
ch = logging.StreamHandler()
# create formatter
formatter = logging.Formatter('%(name)s - %(levelname)s - %(message)s')
# add formatter to ch
# add ch to logger
return logger
logger = create_logger()
def print_usage():
Usage: you can run this script with any valid combination
of the following environment variables (in any order):
KERNEL - the kernel image file to use
BIOS - the bios image file to use
ROOTFS - the rootfs image file or nfsroot directory to use
DEVICE_TREE - the device tree blob to use
MACHINE - the machine name (optional, autodetected from KERNEL filename if unspecified)
Simplified QEMU command-line options can be passed with:
nographic - disable video console
sdl - choose the SDL UI frontend
gtk - choose the Gtk UI frontend
gl - enable virgl-based GL acceleration (also needs gtk or sdl options)
gl-es - enable virgl-based GL acceleration, using OpenGL ES (also needs gtk or sdl options)
egl-headless - enable headless EGL output; use vnc (via publicvnc option) or spice to see it
serial - enable a serial console on /dev/ttyS0
serialstdio - enable a serial console on the console (regardless of graphics mode)
slirp - enable user networking, no root privileges is required
snapshot - don't write changes to back to images
kvm - enable KVM when running x86/x86_64 (VT-capable CPU required)
kvm-vhost - enable KVM with vhost when running x86/x86_64 (VT-capable CPU required)
publicvnc - enable a VNC server open to all hosts
audio - enable audio
[*/]ovmf* - OVMF firmware file or base name for booting with UEFI
tcpserial=<port> - specify tcp serial port number
qemuparams=<xyz> - specify custom parameters to QEMU
bootparams=<xyz> - specify custom kernel parameters during boot
help, -h, --help: print this text
-d, --debug: Enable debug output
-q, --quiet: Hide most output except error messages
runqemu qemuarm
runqemu tmp/deploy/images/qemuarm
runqemu tmp/deploy/images/qemux86/<qemuboot.conf>
runqemu qemux86-64 core-image-sato ext4
runqemu qemux86-64 wic-image-minimal wic
runqemu path/to/bzImage-qemux86.bin path/to/nfsrootdir/ serial
runqemu qemux86 iso/hddimg/wic.vmdk/wic.qcow2/wic.vdi/ramfs/cpio.gz...
runqemu qemux86 qemuparams="-m 256"
runqemu qemux86 bootparams="psplash=false"
runqemu path/to/<image>-<machine>.wic
runqemu path/to/<image>-<machine>.wic.vmdk
def check_tun():
"""Check /dev/net/tun"""
dev_tun = '/dev/net/tun'
if not os.path.exists(dev_tun):
raise RunQemuError("TUN control device %s is unavailable; you may need to enable TUN (e.g. sudo modprobe tun)" % dev_tun)
if not os.access(dev_tun, os.W_OK):
raise RunQemuError("TUN control device %s is not writable, please fix (e.g. sudo chmod 666 %s)" % (dev_tun, dev_tun))
def get_first_file(cmds):
"""Return first file found in wildcard cmds"""
for cmd in cmds:
all_files = glob.glob(cmd)
if all_files:
for f in all_files:
if not os.path.isdir(f):
return f
return ''
class BaseConfig(object):
def __init__(self):
# The self.d saved vars from self.set(), part of them are from qemuboot.conf
self.d = {'QB_KERNEL_ROOT': '/dev/vda'}
# Supported env vars, add it here if a var can be got from env,
# and don't use os.getenv in the code.
self.env_vars = ('MACHINE',
self.qemu_opt = ''
self.qemu_opt_script = ''
self.qemuparams = ''
self.clean_nfs_dir = False
self.nfs_server = ''
self.rootfs = ''
# File name(s) of a OVMF firmware file or variable store,
# to be added with -drive if=pflash.
# Found in the same places as the rootfs, with or without one of
# these suffices: qcow2, bin.
self.ovmf_bios = []
# When enrolling default Secure Boot keys, the hypervisor
# must provide the Platform Key and the first Key Exchange Key
# certificate in the Type 11 SMBIOS table.
self.ovmf_secboot_pkkek1 = ''
self.qemuboot = ''
self.qbconfload = False
self.kernel = ''
self.bios = ''
self.kernel_cmdline = ''
self.kernel_cmdline_script = ''
self.bootparams = ''
self.dtb = ''
self.fstype = ''
self.kvm_enabled = False
self.vhost_enabled = False
self.slirp_enabled = False
self.net_bridge = None
self.nfs_instance = 0
self.nfs_running = False
self.serialconsole = False
self.serialstdio = False
self.cleantap = False
self.saved_stty = ''
self.audio_enabled = False
self.tcpserial_portnum = ''
self.taplock = ''
self.taplock_descriptor = None
self.portlocks = {}
self.bitbake_e = ''
self.snapshot = False
self.wictypes = ('wic', 'wic.vmdk', 'wic.qcow2', 'wic.vdi')
self.fstypes = ('ext2', 'ext3', 'ext4', 'jffs2', 'nfs', 'btrfs',
'cpio.gz', 'cpio', 'ramfs', 'tar.bz2', 'tar.gz')
self.vmtypes = ('hddimg', 'iso')
self.fsinfo = {}
self.network_device = "-device e1000,netdev=net0,mac=@MAC@"
self.cmdline_ip_slirp = "ip=dhcp"
self.cmdline_ip_tap = "ip=192.168.7.@CLIENT@::192.168.7.@GATEWAY@:"
# Use different mac section for tap and slirp to avoid
# conflicts, e.g., when one is running with tap, the other is
# running with slirp.
# The last section is dynamic, which is for avoiding conflicts,
# when multiple qemus are running, e.g., when multiple tap or
# slirp qemus are running.
self.mac_tap = "52:54:00:12:34:"
self.mac_slirp = "52:54:00:12:35:"
# pid of the actual qemu process
self.qemupid = None
# avoid cleanup twice
self.cleaned = False
def acquire_taplock(self, error=True):
logger.debug("Acquiring lockfile %s..." % self.taplock)
self.taplock_descriptor = open(self.taplock, 'w')
fcntl.flock(self.taplock_descriptor, fcntl.LOCK_EX|fcntl.LOCK_NB)
except Exception as e:
msg = "Acquiring lockfile %s failed: %s" % (self.taplock, e)
if error:
if self.taplock_descriptor:
self.taplock_descriptor = None
return False
return True
def release_taplock(self):
if self.taplock_descriptor:
logger.debug("Releasing lockfile for tap device '%s'" % self.tap)
fcntl.flock(self.taplock_descriptor, fcntl.LOCK_UN)
self.taplock_descriptor = None
def check_free_port(self, host, port, lockdir):
""" Check whether the port is free or not """
import socket
from contextlib import closing
lockfile = os.path.join(lockdir, str(port) + '.lock')
if self.acquire_portlock(lockfile):
with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
if sock.connect_ex((host, port)) == 0:
# Port is open, so not free
return False
# Port is not open, so free
return True
return False
def acquire_portlock(self, lockfile):
logger.debug("Acquiring lockfile %s..." % lockfile)
portlock_descriptor = open(lockfile, 'w')
self.portlocks.update({lockfile: portlock_descriptor})
fcntl.flock(self.portlocks[lockfile], fcntl.LOCK_EX|fcntl.LOCK_NB)
except Exception as e:
msg = "Acquiring lockfile %s failed: %s" % (lockfile, e)
if lockfile in self.portlocks.keys() and self.portlocks[lockfile]:
del self.portlocks[lockfile]
return False
return True
def release_portlock(self, lockfile=None):
if lockfile != None:
logger.debug("Releasing lockfile '%s'" % lockfile)
fcntl.flock(self.portlocks[lockfile], fcntl.LOCK_UN)
del self.portlocks[lockfile]
elif len(self.portlocks):
for lockfile, descriptor in self.portlocks.items():
logger.debug("Releasing lockfile '%s'" % lockfile)
fcntl.flock(descriptor, fcntl.LOCK_UN)
self.portlocks = {}
def get(self, key):
if key in self.d:
return self.d.get(key)
elif os.getenv(key):
return os.getenv(key)
return ''
def set(self, key, value):
self.d[key] = value
def is_deploy_dir_image(self, p):
if os.path.isdir(p):
if not'.qemuboot.conf$', '\n'.join(os.listdir(p)), re.M):
logger.debug("Can't find required *.qemuboot.conf in %s" % p)
return False
if not any(map(lambda name: '-image-' in name, os.listdir(p))):
logger.debug("Can't find *-image-* in %s" % p)
return False
return True
return False
def check_arg_fstype(self, fst):
"""Check and set FSTYPE"""
if fst not in self.fstypes + self.vmtypes + self.wictypes:
logger.warning("Maybe unsupported FSTYPE: %s" % fst)
if not self.fstype or self.fstype == fst:
if fst == 'ramfs':
fst = 'cpio.gz'
if fst in ('tar.bz2', 'tar.gz'):
fst = 'nfs'
self.fstype = fst
raise RunQemuError("Conflicting: FSTYPE %s and %s" % (self.fstype, fst))
def set_machine_deploy_dir(self, machine, deploy_dir_image):
logger.debug('MACHINE: %s' % machine)
self.set("MACHINE", machine)
logger.debug('DEPLOY_DIR_IMAGE: %s' % deploy_dir_image)
self.set("DEPLOY_DIR_IMAGE", deploy_dir_image)
def check_arg_nfs(self, p):
if os.path.isdir(p):
self.rootfs = p
m = re.match('(.*):(.*)', p)
self.nfs_server =
self.rootfs =
def check_arg_path(self, p):
- Check whether it is <image>.qemuboot.conf or contains <image>.qemuboot.conf
- Check whether is a kernel file
- Check whether is a image file
- Check whether it is a nfs dir
- Check whether it is a OVMF flash file
if p.endswith('.qemuboot.conf'):
self.qemuboot = p
self.qbconfload = True
elif'\.bin$', p) or'bzImage', p) or \'zImage', p) or'vmlinux', p) or \'fitImage', p) or'uImage', p):
self.kernel = p
elif os.path.exists(p) and (not os.path.isdir(p)) and '-image-' in os.path.basename(p):
self.rootfs = p
# Check filename against self.fstypes can hanlde <file>.cpio.gz,
# otherwise, its type would be "gz", which is incorrect.
fst = ""
for t in self.fstypes:
if p.endswith(t):
fst = t
if not fst:
m ='.*\.(.*)$', self.rootfs)
if m:
fst =
if fst:
qb = re.sub('\.' + fst + "$", '', self.rootfs)
qb = '%s%s' % (re.sub('\.rootfs$', '', qb), '.qemuboot.conf')
if os.path.exists(qb):
self.qemuboot = qb
self.qbconfload = True
logger.warning("%s doesn't exist" % qb)
raise RunQemuError("Can't find FSTYPE from: %s" % p)
elif os.path.isdir(p) or':', p) and'/', p):
if self.is_deploy_dir_image(p):
logger.debug('DEPLOY_DIR_IMAGE: %s' % p)
self.set("DEPLOY_DIR_IMAGE", p)
logger.debug("Assuming %s is an nfs rootfs" % p)
elif os.path.basename(p).startswith('ovmf'):
raise RunQemuError("Unknown path arg %s" % p)
def check_arg_machine(self, arg):
"""Check whether it is a machine"""
if self.get('MACHINE') == arg:
elif self.get('MACHINE') and self.get('MACHINE') != arg:
raise RunQemuError("Maybe conflicted MACHINE: %s vs %s" % (self.get('MACHINE'), arg))
elif'/', arg):
raise RunQemuError("Unknown arg: %s" % arg)
logger.debug('Assuming MACHINE = %s' % arg)
# if we're running under testimage, or similarly as a child
# of an existing bitbake invocation, we can't invoke bitbake
# to validate the MACHINE setting and must assume it's correct...
# FIXME: testimage.bbclass exports these two variables into env,
# are there other scenarios in which we need to support being
# invoked by bitbake?
deploy = self.get('DEPLOY_DIR_IMAGE')
bbchild = deploy and self.get('OE_TMPDIR')
if bbchild:
self.set_machine_deploy_dir(arg, deploy)
# also check whether we're running under a sourced toolchain
# environment file
self.set("MACHINE", arg)
self.bitbake_e = self.run_bitbake_env(arg)
# bitbake -e doesn't report invalid MACHINE as an error, so
# let's check DEPLOY_DIR_IMAGE to make sure that it is a valid
s ='^DEPLOY_DIR_IMAGE="(.*)"', self.bitbake_e, re.M)
if s:
deploy_dir_image =
raise RunQemuError("bitbake -e %s" % self.bitbake_e)
if self.is_deploy_dir_image(deploy_dir_image):
self.set_machine_deploy_dir(arg, deploy_dir_image)
logger.error("%s not a directory valid DEPLOY_DIR_IMAGE" % deploy_dir_image)
self.set("MACHINE", arg)
def set_dri_path(self):
# As runqemu can be run within bitbake (when using testimage, for example),
# we need to ensure that we run host pkg-config, and that it does not
# get mis-directed to native build paths set by bitbake.
del os.environ['PKG_CONFIG_PATH']
del os.environ['PKG_CONFIG_DIR']
del os.environ['PKG_CONFIG_LIBDIR']
del os.environ['PKG_CONFIG_SYSROOT_DIR']
except KeyError:
dripath = subprocess.check_output("PATH=/bin:/usr/bin:$PATH pkg-config --variable=dridriverdir dri", shell=True)
except subprocess.CalledProcessError as e:
raise RunQemuError("Could not determine the path to dri drivers on the host via pkg-config.\nPlease install Mesa development files (particularly, dri.pc) on the host machine.")
os.environ['LIBGL_DRIVERS_PATH'] = dripath.decode('utf-8').strip()
def check_args(self):
for debug in ("-d", "--debug"):
if debug in sys.argv:
for quiet in ("-q", "--quiet"):
if quiet in sys.argv:
if 'gl' not in sys.argv[1:] and 'gl-es' not in sys.argv[1:]:
os.environ['SDL_RENDER_DRIVER'] = 'software'
unknown_arg = ""
for arg in sys.argv[1:]:
if arg in self.fstypes + self.vmtypes + self.wictypes:
elif arg == 'nographic':
self.qemu_opt_script += ' -nographic'
self.kernel_cmdline_script += ' console=ttyS0'
elif arg == 'sdl':
if 'gl' in sys.argv[1:]:
self.qemu_opt_script += ' -vga virtio -display sdl,gl=on,show-cursor=on'
elif 'gl-es' in sys.argv[1:]:
self.qemu_opt_script += ' -vga virtio -display sdl,gl=es,show-cursor=on'
self.qemu_opt_script += ' -display sdl,show-cursor=on'
elif arg == 'gtk':
if 'gl' in sys.argv[1:]:
self.qemu_opt_script += ' -vga virtio -display gtk,gl=on,show-cursor=on'
elif 'gl-es' in sys.argv[1:]:
self.qemu_opt_script += ' -vga virtio -display gtk,gl=es,show-cursor=on'
self.qemu_opt_script += ' -display gtk,show-cursor=on'
elif arg == 'gl' or arg == 'gl-es':
# These args are handled inside sdl or gtk blocks above
elif arg == 'egl-headless':
self.qemu_opt_script += ' -vga virtio -display egl-headless,show-cursor=on'
elif arg == 'serial':
self.kernel_cmdline_script += ' console=ttyS0'
self.serialconsole = True
elif arg == "serialstdio":
self.kernel_cmdline_script += ' console=ttyS0'
self.serialstdio = True
elif arg == 'audio':"Enabling audio in qemu")"Please install sound drivers in linux host")
self.audio_enabled = True
elif arg == 'kvm':
self.kvm_enabled = True
elif arg == 'kvm-vhost':
self.vhost_enabled = True
elif arg == 'slirp':
self.slirp_enabled = True
elif arg.startswith('bridge='):
self.net_bridge = '%s' % arg[len('bridge='):]
elif arg == 'snapshot':
self.snapshot = True
elif arg == 'publicvnc':
self.qemu_opt_script += ' -vnc :0'
elif arg.startswith('tcpserial='):
self.tcpserial_portnum = '%s' % arg[len('tcpserial='):]
elif arg.startswith('qemuparams='):
self.qemuparams = ' %s' % arg[len('qemuparams='):]
elif arg.startswith('bootparams='):
self.bootparams = arg[len('bootparams='):]
elif os.path.exists(arg) or (':', arg) and'/', arg)):
elif'-image-|-image$', arg):
# Lazy rootfs
self.rootfs = arg
elif arg.startswith('ovmf'):
# At last, assume it is the MACHINE
if (not unknown_arg) or unknown_arg == arg:
unknown_arg = arg
raise RunQemuError("Can't handle two unknown args: %s %s\n"
"Try 'runqemu help' on how to use it" % \
(unknown_arg, arg))
# Check to make sure it is a valid machine
if unknown_arg and self.get('MACHINE') != unknown_arg:
if self.get('DEPLOY_DIR_IMAGE'):
machine = os.path.basename(self.get('DEPLOY_DIR_IMAGE'))
if unknown_arg == machine:
self.set("MACHINE", machine)
if not (self.get('DEPLOY_DIR_IMAGE') or self.qbconfload):
s ='^DEPLOY_DIR_IMAGE="(.*)"', self.bitbake_e, re.M)
if s:
def check_kvm(self):
"""Check kvm and kvm-host"""
if not (self.kvm_enabled or self.vhost_enabled):
self.qemu_opt_script += ' %s %s' % (self.get('QB_MACHINE'), self.get('QB_CPU'))
if not self.get('QB_CPU_KVM'):
raise RunQemuError("QB_CPU_KVM is NULL, this board doesn't support kvm")
self.qemu_opt_script += ' %s %s' % (self.get('QB_MACHINE'), self.get('QB_CPU_KVM'))
yocto_kvm_wiki = ""
yocto_paravirt_kvm_wiki = ""
dev_kvm = '/dev/kvm'
dev_vhost = '/dev/vhost-net'
if self.qemu_system.endswith(('i386', 'x86_64')):
with open('/proc/cpuinfo', 'r') as f:
kvm_cap ='vmx|svm', "".join(f.readlines()))
if not kvm_cap:
logger.error("You are trying to enable KVM on a cpu without VT support.")
logger.error("Remove kvm from the command-line, or refer:")
raise RunQemuError(yocto_kvm_wiki)
if not os.path.exists(dev_kvm):
logger.error("Missing KVM device. Have you inserted kvm modules?")
logger.error("For further help see:")
raise RunQemuError(yocto_kvm_wiki)
if os.access(dev_kvm, os.W_OK|os.R_OK):
self.qemu_opt_script += ' -enable-kvm'
if self.get('MACHINE') == "qemux86":
# Workaround for broken APIC window on pre 4.15 host kernels which causes boot hangs
# See YOCTO #12301
# On 64 bit we use x2apic
self.kernel_cmdline_script += " clocksource=kvm-clock hpet=disable noapic nolapic"
logger.error("You have no read or write permission on /dev/kvm.")
logger.error("Please change the ownership of this file as described at:")
raise RunQemuError(yocto_kvm_wiki)
if self.vhost_enabled:
if not os.path.exists(dev_vhost):
logger.error("Missing virtio net device. Have you inserted vhost-net module?")
logger.error("For further help see:")
raise RunQemuError(yocto_paravirt_kvm_wiki)
if not os.access(dev_kvm, os.W_OK|os.R_OK):
logger.error("You have no read or write permission on /dev/vhost-net.")
logger.error("Please change the ownership of this file as described at:")
raise RunQemuError(yocto_kvm_wiki)
def check_fstype(self):
"""Check and setup FSTYPE"""
if not self.fstype:
fstype = self.get('QB_DEFAULT_FSTYPE')
if fstype:
self.fstype = fstype
raise RunQemuError("FSTYPE is NULL!")
# parse QB_FSINFO into dict, e.g. { 'wic': ['no-kernel-in-fs', 'a-flag'], 'ext4': ['another-flag']}
wic_fs = False
qb_fsinfo = self.get('QB_FSINFO')
if qb_fsinfo:
qb_fsinfo = qb_fsinfo.split()
for fsinfo in qb_fsinfo:
fstype, fsflag = fsinfo.split(':')
if fstype == 'wic':
if fsflag == 'no-kernel-in-fs':
wic_fs = True
elif fsflag == 'kernel-in-fs':
wic_fs = False
logger.warn('Unknown flag "%s:%s" in QB_FSINFO', fstype, fsflag)
logger.warn('QB_FSINFO is not supported for image type "%s"', fstype)
if fstype in self.fsinfo:
self.fsinfo[fstype] = [fsflag]
except Exception:
logger.error('Invalid parameter "%s" in QB_FSINFO', fsinfo)
# treat wic images as vmimages (with kernel) or as fsimages (rootfs only)
if wic_fs:
self.fstypes = self.fstypes + self.wictypes
self.vmtypes = self.vmtypes + self.wictypes
def check_rootfs(self):
"""Check and set rootfs"""
if self.fstype == "none":
if self.get('ROOTFS'):
if not self.rootfs:
self.rootfs = self.get('ROOTFS')
elif self.get('ROOTFS') != self.rootfs:
raise RunQemuError("Maybe conflicted ROOTFS: %s vs %s" % (self.get('ROOTFS'), self.rootfs))
if self.fstype == 'nfs':
if self.rootfs and not os.path.exists(self.rootfs):
# Lazy rootfs
self.rootfs = "%s/%s-%s.%s" % (self.get('DEPLOY_DIR_IMAGE'),
self.rootfs, self.get('MACHINE'),
elif not self.rootfs:
cmd_name = '%s/%s*.%s' % (self.get('DEPLOY_DIR_IMAGE'), self.get('IMAGE_NAME'), self.fstype)
cmd_link = '%s/%s*.%s' % (self.get('DEPLOY_DIR_IMAGE'), self.get('IMAGE_LINK_NAME'), self.fstype)
cmds = (cmd_name, cmd_link)
self.rootfs = get_first_file(cmds)
if not self.rootfs:
raise RunQemuError("Failed to find rootfs: %s or %s" % cmds)
if not os.path.exists(self.rootfs):
raise RunQemuError("Can't find rootfs: %s" % self.rootfs)
def setup_pkkek1(self):
Extract from PEM certificate the Platform Key and first Key
Exchange Key certificate string. The hypervisor needs to provide
it in the Type 11 SMBIOS table
pemcert = '%s/%s' % (self.get('DEPLOY_DIR_IMAGE'), 'OvmfPkKek1.pem')
with open(pemcert, 'r') as pemfile:
key ='\n', ''). \
replace('-----BEGIN CERTIFICATE-----', ''). \
replace('-----END CERTIFICATE-----', '')
self.ovmf_secboot_pkkek1 = key
except FileNotFoundError:
raise RunQemuError("Can't open PEM certificate %s " % pemcert)
def check_ovmf(self):
"""Check and set full path for OVMF firmware and variable file(s)."""
for index, ovmf in enumerate(self.ovmf_bios):
if os.path.exists(ovmf):
for suffix in ('qcow2', 'bin'):
path = '%s/%s.%s' % (self.get('DEPLOY_DIR_IMAGE'), ovmf, suffix)
if os.path.exists(path):
self.ovmf_bios[index] = path
if ovmf.endswith('secboot'):
raise RunQemuError("Can't find OVMF firmware: %s" % ovmf)
def check_kernel(self):
"""Check and set kernel"""
# The vm image doesn't need a kernel
if self.fstype in self.vmtypes:
# See if the user supplied a KERNEL option
if self.get('KERNEL'):
self.kernel = self.get('KERNEL')
# QB_DEFAULT_KERNEL is always a full file path
kernel_name = os.path.basename(self.get('QB_DEFAULT_KERNEL'))
# The user didn't want a kernel to be loaded
if kernel_name == "none" and not self.kernel:
deploy_dir_image = self.get('DEPLOY_DIR_IMAGE')
if not self.kernel:
kernel_match_name = "%s/%s" % (deploy_dir_image, kernel_name)
kernel_match_link = "%s/%s" % (deploy_dir_image, self.get('KERNEL_IMAGETYPE'))
kernel_startswith = "%s/%s*" % (deploy_dir_image, self.get('KERNEL_IMAGETYPE'))
cmds = (kernel_match_name, kernel_match_link, kernel_startswith)
self.kernel = get_first_file(cmds)
if not self.kernel:
raise RunQemuError('KERNEL not found: %s, %s or %s' % cmds)
if not os.path.exists(self.kernel):
raise RunQemuError("KERNEL %s not found" % self.kernel)
def check_dtb(self):
"""Check and set dtb"""
# Did the user specify a device tree?
if self.get('DEVICE_TREE'):
self.dtb = self.get('DEVICE_TREE')
if not os.path.exists(self.dtb):
raise RunQemuError('Specified DTB not found: %s' % self.dtb)
dtb = self.get('QB_DTB')
if dtb:
deploy_dir_image = self.get('DEPLOY_DIR_IMAGE')
cmd_match = "%s/%s" % (deploy_dir_image, dtb)
cmd_startswith = "%s/%s*" % (deploy_dir_image, dtb)
cmd_wild = "%s/*.dtb" % deploy_dir_image
cmds = (cmd_match, cmd_startswith, cmd_wild)
self.dtb = get_first_file(cmds)
if not os.path.exists(self.dtb):
raise RunQemuError('DTB not found: %s, %s or %s' % cmds)
def check_bios(self):
"""Check and set bios"""
# See if the user supplied a BIOS option
if self.get('BIOS'):
self.bios = self.get('BIOS')
# QB_DEFAULT_BIOS is always a full file path
bios_name = os.path.basename(self.get('QB_DEFAULT_BIOS'))
# The user didn't want a bios to be loaded
if (bios_name == "" or bios_name == "none") and not self.bios:
if not self.bios:
deploy_dir_image = self.get('DEPLOY_DIR_IMAGE')
self.bios = "%s/%s" % (deploy_dir_image, bios_name)
if not self.bios:
raise RunQemuError('BIOS not found: %s' % bios_match_name)
if not os.path.exists(self.bios):
raise RunQemuError("KERNEL %s not found" % self.bios)
def check_mem(self):
Both qemu and kernel needs memory settings, so check QB_MEM and set it
for both.
s ='-m +([0-9]+)', self.qemuparams)
if s:
self.set('QB_MEM', '-m %s' %
elif not self.get('QB_MEM'):'QB_MEM is not set, use 256M by default')
self.set('QB_MEM', '-m 256')
# Check and remove M or m suffix
qb_mem = self.get('QB_MEM')
if qb_mem.endswith('M') or qb_mem.endswith('m'):
qb_mem = qb_mem[:-1]
# Add -m prefix it not present
if not qb_mem.startswith('-m'):
qb_mem = '-m %s' % qb_mem
self.set('QB_MEM', qb_mem)
mach = self.get('MACHINE')
if not mach.startswith('qemumips'):
self.kernel_cmdline_script += ' mem=%s' % self.get('QB_MEM').replace('-m','').strip() + 'M'
self.qemu_opt_script += ' %s' % self.get('QB_MEM')
def check_tcpserial(self):
if self.tcpserial_portnum:
ports = self.tcpserial_portnum.split(':')
port = ports[0]
if self.get('QB_TCPSERIAL_OPT'):
self.qemu_opt_script += ' ' + self.get('QB_TCPSERIAL_OPT').replace('@PORT@', port)
self.qemu_opt_script += ' -serial tcp:' % port
if len(ports) > 1:
for port in ports[1:]:
self.qemu_opt_script += ' -serial tcp:' % port
def check_and_set(self):
"""Check configs sanity and set when needed"""
if not self.slirp_enabled and not self.net_bridge:
# Check audio
if self.audio_enabled:
if not self.get('QB_AUDIO_DRV'):
raise RunQemuError("QB_AUDIO_DRV is NULL, this board doesn't support audio")
if not self.get('QB_AUDIO_OPT'):
logger.warning('QB_AUDIO_OPT is NULL, you may need define it to make audio work')
self.qemu_opt_script += ' %s' % self.get('QB_AUDIO_OPT')
os.putenv('QEMU_AUDIO_DRV', self.get('QB_AUDIO_DRV'))
os.putenv('QEMU_AUDIO_DRV', 'none')
def read_qemuboot(self):
if not self.qemuboot:
if self.get('DEPLOY_DIR_IMAGE'):
deploy_dir_image = self.get('DEPLOY_DIR_IMAGE')
logger.warning("Can't find qemuboot conf file, DEPLOY_DIR_IMAGE is NULL!")
if self.rootfs and not os.path.exists(self.rootfs):
# Lazy rootfs
machine = self.get('MACHINE')
if not machine:
machine = os.path.basename(deploy_dir_image)
self.qemuboot = "%s/%s-%s.qemuboot.conf" % (deploy_dir_image,
self.rootfs, machine)
cmd = 'ls -t %s/*.qemuboot.conf' % deploy_dir_image
logger.debug('Running %s...' % cmd)
qbs = subprocess.check_output(cmd, shell=True).decode('utf-8')
except subprocess.CalledProcessError as err:
raise RunQemuError(err)
if qbs:
for qb in qbs.split():
# Don't use initramfs when other choices unless fstype is ramfs
if '-initramfs-' in os.path.basename(qb) and self.fstype != 'cpio.gz':
self.qemuboot = qb
if not self.qemuboot:
# Use the first one when no choice
self.qemuboot = qbs.split()[0]
self.qbconfload = True
if not self.qemuboot:
# If we haven't found a .qemuboot.conf at this point it probably
# doesn't exist, continue without
if not os.path.exists(self.qemuboot):
raise RunQemuError("Failed to find %s (wrong image name or BSP does not support running under qemu?)." % self.qemuboot)
logger.debug('CONFFILE: %s' % self.qemuboot)
cf = configparser.ConfigParser()
for k, v in cf.items('config_bsp'):
k_upper = k.upper()
if v.startswith("../"):
v = os.path.abspath(os.path.dirname(self.qemuboot) + "/" + v)
elif v == ".":
v = os.path.dirname(self.qemuboot)
self.set(k_upper, v)
def validate_paths(self):
"""Ensure all relevant path variables are set"""
# When we're started with a *.qemuboot.conf arg assume that image
# artefacts are relative to that file, rather than in whatever
# directory DEPLOY_DIR_IMAGE in the conf file points to.
if self.qbconfload:
imgdir = os.path.realpath(os.path.dirname(self.qemuboot))
if imgdir != os.path.realpath(self.get('DEPLOY_DIR_IMAGE')):'Setting DEPLOY_DIR_IMAGE to folder containing %s (%s)' % (self.qemuboot, imgdir))
self.set('DEPLOY_DIR_IMAGE', imgdir)
# If the STAGING_*_NATIVE directories from the config file don't exist
# and we're in a sourced OE build directory try to extract the paths
# from `bitbake -e`
havenative = os.path.exists(self.get('STAGING_DIR_NATIVE')) and \
if not havenative:
if not self.bitbake_e:
if self.bitbake_e:
native_vars = ['STAGING_DIR_NATIVE']
for nv in native_vars:
s ='^%s="(.*)"' % nv, self.bitbake_e, re.M)
if s and != self.get(nv):'Overriding conf file setting of %s to %s from Bitbake environment' % (nv,
# when we're invoked from a running bitbake instance we won't
# be able to call `bitbake -e`, then try:
# - get OE_TMPDIR from environment and guess paths based on it
# - get OECORE_NATIVE_SYSROOT from environment (for sdk)
tmpdir = self.get('OE_TMPDIR')
oecore_native_sysroot = self.get('OECORE_NATIVE_SYSROOT')
if tmpdir:'Setting STAGING_DIR_NATIVE and STAGING_BINDIR_NATIVE relative to OE_TMPDIR (%s)' % tmpdir)
hostos, _, _, _, machine = os.uname()
buildsys = '%s-%s' % (machine, hostos.lower())
staging_dir_native = '%s/sysroots/%s' % (tmpdir, buildsys)
self.set('STAGING_DIR_NATIVE', staging_dir_native)
elif oecore_native_sysroot:'Setting STAGING_DIR_NATIVE to OECORE_NATIVE_SYSROOT (%s)' % oecore_native_sysroot)
self.set('STAGING_DIR_NATIVE', oecore_native_sysroot)
if self.get('STAGING_DIR_NATIVE'):
# we have to assume that STAGING_BINDIR_NATIVE is at usr/bin
staging_bindir_native = '%s/usr/bin' % self.get('STAGING_DIR_NATIVE')'Setting STAGING_BINDIR_NATIVE to %s' % staging_bindir_native)
self.set('STAGING_BINDIR_NATIVE', '%s/usr/bin' % self.get('STAGING_DIR_NATIVE'))
def print_config(self):
logoutput = ['Continuing with the following parameters:']
if not self.fstype in self.vmtypes:
logoutput.append('KERNEL: [%s]' % self.kernel)
if self.bios:
logoutput.append('BIOS: [%s]' % self.bios)
if self.dtb:
logoutput.append('DTB: [%s]' % self.dtb)
logoutput.append('MACHINE: [%s]' % self.get('MACHINE'))
fstype_flags = ' (' + ', '.join(self.fsinfo[self.fstype]) + ')'
except KeyError:
fstype_flags = ''
logoutput.append('FSTYPE: [%s%s]' % (self.fstype, fstype_flags))
if self.fstype == 'nfs':
logoutput.append('NFS_DIR: [%s]' % self.rootfs)
logoutput.append('ROOTFS: [%s]' % self.rootfs)
if self.ovmf_bios:
logoutput.append('OVMF: %s' % self.ovmf_bios)
if (self.ovmf_secboot_pkkek1):
logoutput.append('SECBOOT PKKEK1: [%s...]' % self.ovmf_secboot_pkkek1[0:100])
logoutput.append('CONFFILE: [%s]' % self.qemuboot)
def setup_nfs(self):
if not self.nfs_server:
if self.slirp_enabled:
self.nfs_server = ''
self.nfs_server = ''
# Figure out a new nfs_instance to allow multiple qemus running.
ps = subprocess.check_output(("ps", "auxww")).decode('utf-8')
pattern = '/bin/unfsd .* -i .*\.pid -e .*/exports([0-9]+) '
all_instances = re.findall(pattern, ps, re.M)
if all_instances:
self.nfs_instance = int(all_instances.pop()) + 1
nfsd_port = 3049 + 2 * self.nfs_instance
mountd_port = 3048 + 2 * self.nfs_instance
# Export vars for runqemu-export-rootfs
export_dict = {
'NFS_INSTANCE': self.nfs_instance,
'NFSD_PORT': nfsd_port,
'MOUNTD_PORT': mountd_port,
for k, v in export_dict.items():
# Use '%s' since they are integers
os.putenv(k, '%s' % v)
self.unfs_opts="nfsvers=3,port=%s,tcp,mountport=%s" % (nfsd_port, mountd_port)
# Extract .tar.bz2 or if no nfs dir
if not (self.rootfs and os.path.isdir(self.rootfs)):
src_prefix = '%s/%s' % (self.get('DEPLOY_DIR_IMAGE'), self.get('IMAGE_LINK_NAME'))
dest = "%s-nfsroot" % src_prefix
if os.path.exists('%s.pseudo_state' % dest):'Use %s as NFS_DIR' % dest)
self.rootfs = dest
src = ""
src1 = '%s.tar.bz2' % src_prefix
src2 = '%s.tar.gz' % src_prefix
if os.path.exists(src1):
src = src1
elif os.path.exists(src2):
src = src2
if not src:
raise RunQemuError("No NFS_DIR is set, and can't find %s or %s to extract" % (src1, src2))'NFS_DIR not found, extracting %s to %s' % (src, dest))
cmd = ('runqemu-extract-sdk', src, dest)'Running %s...' % str(cmd))
if != 0:
raise RunQemuError('Failed to run %s' % cmd)
self.clean_nfs_dir = True
self.rootfs = dest
# Start the userspace NFS server
cmd = ('runqemu-export-rootfs', 'start', self.rootfs)'Running %s...' % str(cmd))
if != 0:
raise RunQemuError('Failed to run %s' % cmd)
self.nfs_running = True
def setup_net_bridge(self):
self.set('NETWORK_CMD', '-netdev bridge,br=%s,id=net0,helper=%s -device virtio-net-pci,netdev=net0 ' % (
self.net_bridge, os.path.join(self.bindir_native, 'qemu-oe-bridge-helper')))
def setup_slirp(self):
"""Setup user networking"""
if self.fstype == 'nfs':
netconf = " " + self.cmdline_ip_slirp"Network configuration:%s", netconf)
self.kernel_cmdline_script += netconf
# Port mapping
hostfwd = ",hostfwd=tcp::2222-:22,hostfwd=tcp::2323-:23"
qb_slirp_opt_default = "-netdev user,id=net0%s,tftp=%s" % (hostfwd, self.get('DEPLOY_DIR_IMAGE'))
qb_slirp_opt = self.get('QB_SLIRP_OPT') or qb_slirp_opt_default
# Figure out the port
ports = re.findall('hostfwd=[^-]*:([0-9]+)-[^,-]*', qb_slirp_opt)
ports = [int(i) for i in ports]
mac = 2
lockdir = "/tmp/qemu-port-locks"
if not os.path.exists(lockdir):
# There might be a race issue when multi runqemu processess are
# running at the same time.
os.chmod(lockdir, 0o777)
except FileExistsError:
# Find a free port to avoid conflicts
for p in ports[:]:
p_new = p
while not self.check_free_port('localhost', p_new, lockdir):
p_new += 1
mac += 1
while p_new in ports:
p_new += 1
mac += 1
if p != p_new:
qb_slirp_opt = re.sub(':%s-' % p, ':%s-' % p_new, qb_slirp_opt)"Port forward changed: %s -> %s" % (p, p_new))
mac = "%s%02x" % (self.mac_slirp, mac)
self.set('NETWORK_CMD', '%s %s' % (self.network_device.replace('@MAC@', mac), qb_slirp_opt))
# Print out port foward
hostfwd = re.findall('(hostfwd=[^,]*)', qb_slirp_opt)
if hostfwd:'Port forward: %s' % ' '.join(hostfwd))
def setup_tap(self):
"""Setup tap"""
# This file is created when runqemu-gen-tapdevs creates a bank of tap
# devices, indicating that the user should not bring up new ones using
# sudo.
nosudo_flag = '/etc/runqemu-nosudo'
self.qemuifup = shutil.which('runqemu-ifup')
self.qemuifdown = shutil.which('runqemu-ifdown')
ip = shutil.which('ip')
lockdir = "/tmp/qemu-tap-locks"
if not (self.qemuifup and self.qemuifdown and ip):
logger.error("runqemu-ifup: %s" % self.qemuifup)
logger.error("runqemu-ifdown: %s" % self.qemuifdown)
logger.error("ip: %s" % ip)
raise OEPathError("runqemu-ifup, runqemu-ifdown or ip not found")
if not os.path.exists(lockdir):
# There might be a race issue when multi runqemu processess are
# running at the same time.
os.chmod(lockdir, 0o777)
except FileExistsError:
cmd = (ip, 'link')
logger.debug('Running %s...' % str(cmd))
ip_link = subprocess.check_output(cmd).decode('utf-8')
# Matches line like: 6: tap0: <foo>
possibles = re.findall('^[0-9]+: +(tap[0-9]+): <.*', ip_link, re.M)
tap = ""
for p in possibles:
lockfile = os.path.join(lockdir, p)
if os.path.exists('%s.skip' % lockfile):'Found %s.skip, skipping %s' % (lockfile, p))
self.taplock = lockfile + '.lock'
if self.acquire_taplock(error=False):
tap = p"Using preconfigured tap device %s" % tap)"If this is not intended, touch %s.skip to make runqemu skip %s." %(lockfile, tap))
if not tap:
if os.path.exists(nosudo_flag):
logger.error("Error: There are no available tap devices to use for networking,")
logger.error("and I see %s exists, so I am not going to try creating" % nosudo_flag)
raise RunQemuError("a new one with sudo.")
gid = os.getgid()
uid = os.getuid()"Setting up tap interface under sudo")
cmd = ('sudo', self.qemuifup, str(uid), str(gid), self.bindir_native)
tap = subprocess.check_output(cmd).decode('utf-8').strip()
except subprocess.CalledProcessError as e:
logger.error('Setting up tap device failed:\n%s\nRun runqemu-gen-tapdevs to manually create one.' % str(e))
lockfile = os.path.join(lockdir, tap)
self.taplock = lockfile + '.lock'
self.cleantap = True
logger.debug('Created tap: %s' % tap)
if not tap:
logger.error("Failed to setup tap device. Run runqemu-gen-tapdevs to manually create.")
self.tap = tap
tapnum = int(tap[3:])
gateway = tapnum * 2 + 1
client = gateway + 1
if self.fstype == 'nfs':
netconf = " " + self.cmdline_ip_tap
netconf = netconf.replace('@CLIENT@', str(client))
netconf = netconf.replace('@GATEWAY@', str(gateway))"Network configuration:%s", netconf)
self.kernel_cmdline_script += netconf
mac = "%s%02x" % (self.mac_tap, client)
qb_tap_opt = self.get('QB_TAP_OPT')
if qb_tap_opt:
qemu_tap_opt = qb_tap_opt.replace('@TAP@', tap)
qemu_tap_opt = "-netdev tap,id=net0,ifname=%s,script=no,downscript=no" % (self.tap)
if self.vhost_enabled:
qemu_tap_opt += ',vhost=on'
self.set('NETWORK_CMD', '%s %s' % (self.network_device.replace('@MAC@', mac), qemu_tap_opt))
def setup_network(self):
if self.get('QB_NET') == 'none':
if sys.stdin.isatty():
self.saved_stty = subprocess.check_output(("stty", "-g")).decode('utf-8').strip()
self.network_device = self.get('QB_NETWORK_DEVICE') or self.network_device
if self.net_bridge:
elif self.slirp_enabled:
self.cmdline_ip_slirp = self.get('QB_CMDLINE_IP_SLIRP') or self.cmdline_ip_slirp
self.cmdline_ip_tap = self.get('QB_CMDLINE_IP_TAP') or self.cmdline_ip_tap
def setup_rootfs(self):
if self.get('QB_ROOTFS') == 'none':
if 'wic.' in self.fstype:
self.fstype = self.fstype[4:]
rootfs_format = self.fstype if self.fstype in ('vmdk', 'qcow2', 'vdi') else 'raw'
qb_rootfs_opt = self.get('QB_ROOTFS_OPT')
if qb_rootfs_opt:
self.rootfs_options = qb_rootfs_opt.replace('@ROOTFS@', self.rootfs)
self.rootfs_options = '-drive file=%s,if=virtio,format=%s' % (self.rootfs, rootfs_format)
qb_rootfs_extra_opt = self.get("QB_ROOTFS_EXTRA_OPT")
if qb_rootfs_extra_opt and not qb_rootfs_extra_opt.startswith(","):
qb_rootfs_extra_opt = "," + qb_rootfs_extra_opt
if self.fstype in ('cpio.gz', 'cpio'):
self.kernel_cmdline = 'root=/dev/ram0 rw debugshell'
self.rootfs_options = '-initrd %s' % self.rootfs
vm_drive = ''
if self.fstype in self.vmtypes:
if self.fstype == 'iso':
vm_drive = '-drive file=%s,if=virtio,media=cdrom' % self.rootfs
elif self.get('QB_DRIVE_TYPE'):
drive_type = self.get('QB_DRIVE_TYPE')
if drive_type.startswith("/dev/sd"):'Using scsi drive')
vm_drive = '-drive if=none,id=hd,file=%s,format=%s -device virtio-scsi-pci,id=scsi -device scsi-hd,drive=hd%s' \
% (self.rootfs, rootfs_format, qb_rootfs_extra_opt)
elif drive_type.startswith("/dev/hd"):'Using ide drive')
vm_drive = "-drive file=%s,format=%s" % (self.rootfs, rootfs_format)
elif drive_type.startswith("/dev/vdb"):'Using block virtio drive');
vm_drive = '-drive id=disk0,file=%s,if=none,format=%s -device virtio-blk-device,drive=disk0%s' \
% (self.rootfs, rootfs_format,qb_rootfs_extra_opt)
# virtio might have been selected explicitly (just use it), or
# is used as fallback (then warn about that).
if not drive_type.startswith("/dev/vd"):
logger.warning("Unknown QB_DRIVE_TYPE: %s" % drive_type)
logger.warning("Failed to figure out drive type, consider define or fix QB_DRIVE_TYPE")
logger.warning('Trying to use virtio block drive')
vm_drive = '-drive if=virtio,file=%s,format=%s' % (self.rootfs, rootfs_format)
# All branches above set vm_drive.
self.rootfs_options = vm_drive
if not self.fstype in self.vmtypes:
self.rootfs_options += ' -no-reboot'
self.kernel_cmdline = 'root=%s rw' % (self.get('QB_KERNEL_ROOT'))
if self.fstype == 'nfs':
self.rootfs_options = ''
k_root = '/dev/nfs nfsroot=%s:%s,%s' % (self.nfs_server, os.path.abspath(self.rootfs), self.unfs_opts)
self.kernel_cmdline = 'root=%s rw' % k_root
if self.fstype == 'none':
self.rootfs_options = ''
self.set('ROOTFS_OPTIONS', self.rootfs_options)
def guess_qb_system(self):
"""attempt to determine the appropriate qemu-system binary"""
mach = self.get('MACHINE')
if not mach:
search = '.*(qemux86-64|qemux86|qemuarm64|qemuarm|qemumips64|qemumips64el|qemumipsel|qemumips|qemuppc).*'
if self.rootfs:
match = re.match(search, self.rootfs)
if match:
mach =
elif self.kernel:
match = re.match(search, self.kernel)
if match:
mach =
if not mach:
return None
if mach == 'qemuarm':
qbsys = 'arm'
elif mach == 'qemuarm64':
qbsys = 'aarch64'
elif mach == 'qemux86':
qbsys = 'i386'
elif mach == 'qemux86-64':
qbsys = 'x86_64'
elif mach == 'qemuppc':
qbsys = 'ppc'
elif mach == 'qemumips':
qbsys = 'mips'
elif mach == 'qemumips64':
qbsys = 'mips64'
elif mach == 'qemumipsel':
qbsys = 'mipsel'
elif mach == 'qemumips64el':
qbsys = 'mips64el'
elif mach == 'qemuriscv64':
qbsys = 'riscv64'
elif mach == 'qemuriscv32':
qbsys = 'riscv32'
logger.error("Unable to determine QEMU PC System emulator for %s machine." % mach)
logger.error("As %s is not among valid QEMU machines such as," % mach)
logger.error("qemux86-64, qemux86, qemuarm64, qemuarm, qemumips64, qemumips64el, qemumipsel, qemumips, qemuppc")
raise RunQemuError("Set qb_system_name with suitable QEMU PC System emulator in .*qemuboot.conf.")
return 'qemu-system-%s' % qbsys
def check_qemu_system(self):
qemu_system = self.get('QB_SYSTEM_NAME')
if not qemu_system:
qemu_system = self.guess_qb_system()
if not qemu_system:
raise RunQemuError("Failed to boot, QB_SYSTEM_NAME is NULL!")
self.qemu_system = qemu_system
def setup_final(self):
qemu_bin = os.path.join(self.bindir_native, self.qemu_system)
# It is possible to have qemu-native in ASSUME_PROVIDED, and it won't
# find QEMU in sysroot, it needs to use host's qemu.
if not os.path.exists(qemu_bin):"QEMU binary not found in %s, trying host's QEMU" % qemu_bin)
for path in (os.environ['PATH'] or '').split(':'):
qemu_bin_tmp = os.path.join(path, self.qemu_system)"Trying: %s" % qemu_bin_tmp)
if os.path.exists(qemu_bin_tmp):
qemu_bin = qemu_bin_tmp
if not os.path.isabs(qemu_bin):
qemu_bin = os.path.abspath(qemu_bin)"Using host's QEMU: %s" % qemu_bin)
if not os.access(qemu_bin, os.X_OK):
raise OEPathError("No QEMU binary '%s' could be found" % qemu_bin)
self.qemu_opt = "%s %s %s %s" % (qemu_bin, self.get('NETWORK_CMD'), self.get('ROOTFS_OPTIONS'), self.get('QB_OPT_APPEND'))
for ovmf in self.ovmf_bios:
format = ovmf.rsplit('.', 1)[-1]
self.qemu_opt += ' -drive if=pflash,format=%s,file=%s' % (format, ovmf)
self.qemu_opt += ' ' + self.qemu_opt_script
if self.ovmf_secboot_pkkek1:
# Provide the Platform Key and first Key Exchange Key certificate as an
# OEM string in the SMBIOS Type 11 table. Prepend the certificate string
# with "application prefix" of the EnrollDefaultKeys.efi application
self.qemu_opt += ' -smbios type=11,value=4e32566d-8e9e-4f52-81d3-5bb9715f9727:' \
+ self.ovmf_secboot_pkkek1
# Append qemuparams to override previous settings
if self.qemuparams:
self.qemu_opt += ' ' + self.qemuparams
if self.snapshot:
self.qemu_opt += " -snapshot"
if self.serialconsole:
if sys.stdin.isatty():
subprocess.check_call(("stty", "intr", "^]"))"Interrupt character is '^]'")
first_serial = ""
if not"-nographic", self.qemu_opt):
first_serial = "-serial mon:vc"
# We always want a ttyS1. Since qemu by default adds a serial
# port when nodefaults is not specified, it seems that all that
# would be needed is to make sure a "-serial" is there. However,
# it appears that when "-serial" is specified, it ignores the
# default serial port that is normally added. So here we make
# sure to add two -serial if there are none. And only one if
# there is one -serial already.
serial_num = len(re.findall("-serial", self.qemu_opt))
if serial_num == 0:
self.qemu_opt += " %s %s" % (first_serial, self.get("QB_SERIAL_OPT"))
elif serial_num == 1:
self.qemu_opt += " %s" % self.get("QB_SERIAL_OPT")
# We always wants ttyS0 and ttyS1 in qemu machines (see SERIAL_CONSOLES),
# if not serial or serialtcp options was specified only ttyS0 is created
# and sysvinit shows an error trying to enable ttyS1:
# INIT: Id "S1" respawning too fast: disabled for 5 minutes
serial_num = len(re.findall("-serial", self.qemu_opt))
if serial_num == 0:
if"-nographic", self.qemu_opt) or self.serialstdio:
self.qemu_opt += " -serial mon:stdio -serial null"
self.qemu_opt += " -serial mon:vc -serial null"
def start_qemu(self):
import shlex
if self.kernel:
kernel_opts = "-kernel %s -append '%s %s %s %s'" % (self.kernel, self.kernel_cmdline,
self.kernel_cmdline_script, self.get('QB_KERNEL_CMDLINE_APPEND'),
if self.bios:
kernel_opts += " -bios %s" % self.bios
if self.dtb:
kernel_opts += " -dtb %s" % self.dtb
kernel_opts = ""
cmd = "%s %s" % (self.qemu_opt, kernel_opts)
cmds = shlex.split(cmd)'Running %s\n' % cmd)
pass_fds = []
if self.taplock_descriptor:
pass_fds = [self.taplock_descriptor.fileno()]
if len(self.portlocks):
for descriptor in self.portlocks.values():
process = subprocess.Popen(cmds, stderr=subprocess.PIPE, pass_fds=pass_fds)
self.qemupid =
retcode = process.wait()
if retcode:
if retcode == -signal.SIGTERM:"Qemu terminated by SIGTERM")
logger.error("Failed to run qemu: %s",
def cleanup(self):
if self.cleaned:
# avoid dealing with SIGTERM when cleanup function is running
signal.signal(signal.SIGTERM, signal.SIG_IGN)"Cleaning up")
if self.cleantap:
cmd = ('sudo', self.qemuifdown, self.tap, self.bindir_native)
logger.debug('Running %s' % str(cmd))
if self.nfs_running:"Shutting down the userspace NFS server...")
cmd = ("runqemu-export-rootfs", "stop", self.rootfs)
logger.debug('Running %s' % str(cmd))
if self.saved_stty:
subprocess.check_call(("stty", self.saved_stty))
if self.clean_nfs_dir:'Removing %s' % self.rootfs)
shutil.rmtree('%s.pseudo_state' % self.rootfs)
self.cleaned = True
def run_bitbake_env(self, mach=None):
bitbake = shutil.which('bitbake')
if not bitbake:
if not mach:
mach = self.get('MACHINE')
multiconfig = self.get('MULTICONFIG')
if multiconfig:
multiconfig = "mc:%s" % multiconfig
if mach:
cmd = 'MACHINE=%s bitbake -e %s' % (mach, multiconfig)
cmd = 'bitbake -e %s' % multiconfig'Running %s...' % cmd)
return subprocess.check_output(cmd, shell=True).decode('utf-8')
def load_bitbake_env(self, mach=None):
if self.bitbake_e:
self.bitbake_e = self.run_bitbake_env(mach=mach)
except subprocess.CalledProcessError as err:
self.bitbake_e = ''
logger.warning("Couldn't run 'bitbake -e' to gather environment information:\n%s" % err.output.decode('utf-8'))
def validate_combos(self):
if (self.fstype in self.vmtypes) and self.kernel:
raise RunQemuError("%s doesn't need kernel %s!" % (self.fstype, self.kernel))
def bindir_native(self):
result = self.get('STAGING_BINDIR_NATIVE')
if result and os.path.exists(result):
return result
cmd = ['bitbake', '-e']
multiconfig = self.get('MULTICONFIG')
if multiconfig:
cmd.append('mc:%s:qemu-helper-native' % multiconfig)
cmd.append('qemu-helper-native')'Running %s...' % str(cmd))
out = subprocess.check_output(cmd).decode('utf-8')
match ='^STAGING_BINDIR_NATIVE="(.*)"', out, re.M)
if match:
result =
if os.path.exists(result):
self.set('STAGING_BINDIR_NATIVE', result)
return result
raise RunQemuError("Native sysroot directory %s doesn't exist" % result)
raise RunQemuError("Can't find STAGING_BINDIR_NATIVE in '%s' output" % cmd)
def main():
if "help" in sys.argv or '-h' in sys.argv or '--help' in sys.argv:
return 0
config = BaseConfig()
def sigterm_handler(signum, frame):"SIGTERM received")
os.kill(config.qemupid, signal.SIGTERM)
# Deliberately ignore the return code of 'tput smam'.["tput", "smam"])
signal.signal(signal.SIGTERM, sigterm_handler)
# Check whether the combos is valid or not
except RunQemuError as err:
return 1
except Exception as err:
import traceback
return 1
# Deliberately ignore the return code of 'tput smam'.["tput", "smam"])
if __name__ == "__main__":