blob: 98c3ca68819fac0293fa92e178770ca996319493 [file] [log] [blame]
#!/usr/bin/env python
# TODO: openbmc/openbmc#2994 remove python 2 support
try: # python 2
import gobject
except ImportError: # python 3
from gi.repository import GObject as gobject
import dbus
import dbus.service
import dbus.mainloop.glib
import subprocess
import tempfile
import shutil
import tarfile
import os
from obmc.dbuslib.bindings import get_dbus, DbusProperties, DbusObjectManager
DBUS_NAME = 'org.openbmc.control.BmcFlash'
OBJ_NAME = '/org/openbmc/control/flash/bmc'
DOWNLOAD_INTF = 'org.openbmc.managers.Download'
BMC_DBUS_NAME = 'xyz.openbmc_project.State.BMC'
BMC_OBJ_NAME = '/xyz/openbmc_project/state/bmc0'
UPDATE_PATH = '/run/initramfs'
def doExtract(members, files):
for tarinfo in members:
if tarinfo.name in files:
yield tarinfo
def save_fw_env():
fw_env = "/etc/fw_env.config"
lines = 0
files = []
envcfg = open(fw_env, 'r')
try:
for line in envcfg.readlines():
# ignore lines that are blank or start with #
if (line.startswith("#")):
continue
if (not len(line.strip())):
continue
fn = line.partition("\t")[0]
files.append(fn)
lines += 1
finally:
envcfg.close()
if (lines < 1 or lines > 2 or (lines == 2 and files[0] != files[1])):
raise Exception("Error parsing %s\n" % fw_env)
shutil.copyfile(files[0], os.path.join(UPDATE_PATH, "image-u-boot-env"))
class BmcFlashControl(DbusProperties, DbusObjectManager):
def __init__(self, bus, name):
super(BmcFlashControl, self).__init__(
conn=bus,
object_path=name)
self.Set(DBUS_NAME, "status", "Idle")
self.Set(DBUS_NAME, "filename", "")
self.Set(DBUS_NAME, "preserve_network_settings", True)
self.Set(DBUS_NAME, "restore_application_defaults", False)
self.Set(DBUS_NAME, "update_kernel_and_apps", False)
self.Set(DBUS_NAME, "clear_persistent_files", False)
self.Set(DBUS_NAME, "auto_apply", False)
bus.add_signal_receiver(
self.download_error_handler, signal_name="DownloadError")
bus.add_signal_receiver(
self.download_complete_handler, signal_name="DownloadComplete")
self.update_process = None
self.progress_name = None
@dbus.service.method(
DBUS_NAME, in_signature='ss', out_signature='')
def updateViaTftp(self, ip, filename):
self.Set(DBUS_NAME, "status", "Downloading")
self.TftpDownload(ip, filename)
@dbus.service.method(
DBUS_NAME, in_signature='s', out_signature='')
def update(self, filename):
self.Set(DBUS_NAME, "filename", filename)
self.download_complete_handler(filename, filename)
@dbus.service.signal(DOWNLOAD_INTF, signature='ss')
def TftpDownload(self, ip, filename):
self.Set(DBUS_NAME, "filename", filename)
pass
# Signal handler
def download_error_handler(self, filename):
if (filename == self.Get(DBUS_NAME, "filename")):
self.Set(DBUS_NAME, "status", "Download Error")
def download_complete_handler(self, outfile, filename):
# do update
if (filename != self.Get(DBUS_NAME, "filename")):
return
print("Download complete. Updating...")
self.Set(DBUS_NAME, "status", "Download Complete")
copy_files = {}
# determine needed files
if not self.Get(DBUS_NAME, "update_kernel_and_apps"):
copy_files["image-bmc"] = True
else:
copy_files["image-kernel"] = True
copy_files["image-rofs"] = True
if self.Get(DBUS_NAME, "restore_application_defaults"):
copy_files["image-rwfs"] = True
# make sure files exist in archive
try:
tar = tarfile.open(outfile, "r")
files = {}
for f in tar.getnames():
files[f] = True
tar.close()
for f in list(copy_files.keys()):
if f not in files:
raise Exception(
"ERROR: File not found in update archive: "+f)
except Exception as e:
print(str(e))
self.Set(DBUS_NAME, "status", "Unpack Error")
return
try:
tar = tarfile.open(outfile, "r")
tar.extractall(UPDATE_PATH, members=doExtract(tar, copy_files))
tar.close()
if self.Get(DBUS_NAME, "clear_persistent_files"):
print("Removing persistent files")
try:
os.unlink(UPDATE_PATH+"/whitelist")
except OSError as e:
if (e.errno == errno.EISDIR):
pass
elif (e.errno == errno.ENOENT):
pass
else:
raise
try:
wldir = UPDATE_PATH + "/whitelist.d"
for file in os.listdir(wldir):
os.unlink(os.path.join(wldir, file))
except OSError as e:
if (e.errno == errno.EISDIR):
pass
else:
raise
if self.Get(DBUS_NAME, "preserve_network_settings"):
print("Preserving network settings")
save_fw_env()
except Exception as e:
print(str(e))
self.Set(DBUS_NAME, "status", "Unpack Error")
self.Verify()
def Verify(self):
self.Set(DBUS_NAME, "status", "Checking Image")
try:
subprocess.check_call([
"/run/initramfs/update",
"--no-flash",
"--no-save-files",
"--no-restore-files",
"--no-clean-saved-files"])
self.Set(DBUS_NAME, "status", "Image ready to apply.")
if (self.Get(DBUS_NAME, "auto_apply")):
self.Apply()
except Exception:
self.Set(DBUS_NAME, "auto_apply", False)
try:
subprocess.check_output([
"/run/initramfs/update",
"--no-flash",
"--ignore-mount",
"--no-save-files",
"--no-restore-files",
"--no-clean-saved-files"],
stderr=subprocess.STDOUT)
self.Set(
DBUS_NAME, "status",
"Deferred for mounted filesystem. reboot BMC to apply.")
except subprocess.CalledProcessError as e:
self.Set(
DBUS_NAME, "status", "Verify error: %s" % e.output)
except OSError as e:
self.Set(
DBUS_NAME, "status",
"Verify error: problem calling update: %s" % e.strerror)
def Cleanup(self):
if self.progress_name:
try:
os.unlink(self.progress_name)
self.progress_name = None
except oserror as e:
if e.errno == EEXIST:
pass
raise
self.update_process = None
self.Set(DBUS_NAME, "status", "Idle")
@dbus.service.method(
DBUS_NAME, in_signature='', out_signature='')
def Abort(self):
if self.update_process:
try:
self.update_process.kill()
except Exception:
pass
for file in os.listdir(UPDATE_PATH):
if file.startswith('image-'):
os.unlink(os.path.join(UPDATE_PATH, file))
self.Cleanup()
@dbus.service.method(
DBUS_NAME, in_signature='', out_signature='s')
def GetUpdateProgress(self):
msg = ""
if self.update_process and self.update_process.returncode is None:
self.update_process.poll()
if (self.update_process is None):
pass
elif (self.update_process.returncode > 0):
self.Set(DBUS_NAME, "status", "Apply failed")
elif (self.update_process.returncode is None):
pass
else: # (self.update_process.returncode == 0)
files = ""
for file in os.listdir(UPDATE_PATH):
if file.startswith('image-'):
files = files + file
if files == "":
msg = "Apply Complete. Reboot to take effect."
else:
msg = "Apply Incomplete, Remaining:" + files
self.Set(DBUS_NAME, "status", msg)
msg = self.Get(DBUS_NAME, "status") + "\n"
if self.progress_name:
try:
prog = open(self.progress_name, 'r')
for line in prog:
# strip off initial sets of xxx\r here
# ignore crlf at the end
# cr will be -1 if no '\r' is found
cr = line.rfind("\r", 0, -2)
msg = msg + line[cr + 1:]
except OSError as e:
if (e.error == EEXIST):
pass
raise
return msg
@dbus.service.method(
DBUS_NAME, in_signature='', out_signature='')
def Apply(self):
progress = None
self.Set(DBUS_NAME, "status", "Writing images to flash")
try:
progress = tempfile.NamedTemporaryFile(
delete=False, prefix="progress.")
self.progress_name = progress.name
self.update_process = subprocess.Popen([
"/run/initramfs/update"],
stdout=progress.file,
stderr=subprocess.STDOUT)
except Exception as e:
try:
progress.close()
os.unlink(progress.name)
self.progress_name = None
except Exception:
pass
raise
try:
progress.close()
except Exception:
pass
@dbus.service.method(
DBUS_NAME, in_signature='', out_signature='')
def PrepareForUpdate(self):
subprocess.call([
"fw_setenv",
"openbmconce",
"copy-files-to-ram copy-base-filesystem-to-ram"])
# Set the variable twice so that it is written to both environments of
# the u-boot redundant environment variables since initramfs can only
# read one of the environments.
subprocess.call([
"fw_setenv",
"openbmconce",
"copy-files-to-ram copy-base-filesystem-to-ram"])
self.Set(DBUS_NAME, "status", "Switch to update mode in progress")
o = bus.get_object(BMC_DBUS_NAME, BMC_OBJ_NAME)
intf = dbus.Interface(o, "org.freedesktop.DBus.Properties")
intf.Set(BMC_DBUS_NAME,
"RequestedBMCTransition",
"xyz.openbmc_project.State.BMC.Transition.Reboot")
if __name__ == '__main__':
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
bus = get_dbus()
obj = BmcFlashControl(bus, OBJ_NAME)
mainloop = gobject.MainLoop()
obj.unmask_signals()
name = dbus.service.BusName(DBUS_NAME, bus)
print("Running Bmc Flash Control")
mainloop.run()
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4