path: root/pyflashbmc/bmc_update.py

#!/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