feat(mtda): add command to apply firmware update

mtda-cli

@@ -452,6 +452,37 @@ class Application:
             client.monitor_remote(self.remote, None)
         return result
 
+    def system_cmd(self, args):
+        cmds = {
+           'update': self.system_update
+        }
+
+        return cmds[args.subcommand](args)
+
+    def system_update(self, args=None):
+        result = 0
+        client = self.agent
+        self.imgname = os.path.basename(args.image)
+        # TODO: there is currently no way back!
+        client.storage_to_sysupdate()
+
+        try:
+            client.monitor_remote(self.remote, self.screen)
+
+            client.system_update_image(args.image)
+            sys.stdout.write("\n")
+            sys.stdout.flush()
+        except Exception as e:
+            import traceback
+            traceback.print_exc()
+            msg = e.msg if hasattr(e, 'msg') else str(e)
+            print(f"\n'system update' failed! ({msg})",
+                  file=sys.stderr)
+            result = 1
+        finally:
+            client.monitor_remote(self.remote, None)
+        return result
+
     def target_uptime(self):
         result = ""
         uptime = self.client().target_uptime()
@@ -833,6 +864,25 @@ class Application:
             help="Path to image file"
         )
 
+        cmd = self.system_cmd
+        p = subparsers.add_parser(
+            "system",
+            help="Interact with the mtda system",
+        )
+        p.set_defaults(func=cmd)
+        subsub = p.add_subparsers(dest="subcommand")
+        subsub.required = True
+        s = subsub.add_parser(
+            "update",
+            help="Update the mtda system"
+        )
+        s.add_argument(
+            "image",
+            metavar="image",
+            type=str,
+            help="Path to swu file"
+        )
+
         # subcommand: target
         cmd = self.target_cmd
         p = subparsers.add_parser(

mtda/client.py

@@ -274,6 +274,21 @@ def parseBmap(self, bmap, bmap_path):
             return None
         return bmapDict
 
+    def system_update_image(self, path, callback=None):
+        blksz = self._agent.blksz
+        impl = self._impl
+        session = self._session
+
+        # Get file handler from specified path
+        file = ImageFile.new(path, impl, session, blksz, callback)
+        self.storage_open(file.size)
+        try:
+            file.prepare(self._data, file.size)
+            file.copy()
+            file.flush()
+        finally:
+            self.storage_close()
+
     def start(self):
         return self._agent.start()
 

mtda/main.py

@@ -1041,6 +1041,20 @@ def storage_to_target(self, **kwargs):
         self.mtda.debug(3, f"main.storage_to_target(): {str(result)}")
         return result
 
+    @Pyro4.expose
+    def storage_to_sysupdate(self, **kwags):
+        # TODO: currently there is no way to go back!
+        from mtda.storage.swupdate import SWUpdate
+        from mtda.storage.writer import AsyncImageWriter
+
+        # TODO: we need to overwrite the global storage object,
+        # as internal calls rely on mtda.storage_status()
+        self.storage = SWUpdate(self)
+        self._writer = AsyncImageWriter(self, self.storage)
+        # TODO: this is technically not true, but this value
+        # is checked all over the place
+        self._storage_event(CONSTS.STORAGE.ON_HOST)
+
     @Pyro4.expose
     def storage_swap(self, **kwargs):
         self.mtda.debug(3, "main.storage_swap()")

mtda/storage/helpers/swupdate_ipc.py

@@ -0,0 +1,162 @@
+# ---------------------------------------------------------------------------
+# Helper class for images
+# ---------------------------------------------------------------------------
+#
+# This software is a part of MTDA.
+# Copyright (C) 2025 Siemens AG
+#
+# ---------------------------------------------------------------------------
+# SPDX-License-Identifier: MIT
+# ---------------------------------------------------------------------------
+#
+# The type definitions need to match the upstream protocol definitions in
+# https://github.com/sbabic/swupdate/blob/master/include/network_ipc.h
+
+import ctypes
+
+# Constants
+IPC_MAGIC = 0x14052001
+SWUPDATE_API_VERSION = 0x1
+
+
+# Enums
+class msgtype(ctypes.c_int):
+    REQ_INSTALL = 0
+    ACK = 1
+    NACK = 2
+    GET_STATUS = 3
+    POST_UPDATE = 4
+    SWUPDATE_SUBPROCESS = 5
+    SET_AES_KEY = 6
+    SET_UPDATE_STATE = 7
+    GET_UPDATE_STATE = 8
+    REQ_INSTALL_EXT = 9
+    SET_VERSIONS_RANGE = 10
+    NOTIFY_STREAM = 11
+    GET_HW_REVISION = 12
+    SET_SWUPDATE_VARS = 13
+    GET_SWUPDATE_VARS = 14
+
+
+class CMD_TYPE(ctypes.c_int):
+    CMD_ACTIVATION = 0
+    CMD_CONFIG = 1
+    CMD_ENABLE = 2
+    CMD_GET_STATUS = 3
+    CMD_SET_DOWNLOAD_URL = 4
+
+
+class run_type(ctypes.c_int):
+    RUN_DEFAULT = 0
+    RUN_DRYRUN = 1
+    RUN_INSTALL = 2
+
+
+# Structures
+class sourcetype(ctypes.c_int):
+    SOURCE_UNKNOWN = 0
+    SOURCE_FILE = 1
+    SOURCE_NETWORK = 2
+    SOURCE_USB = 3
+
+
+class swupdate_request(ctypes.Structure):
+    _fields_ = [
+        ("apiversion", ctypes.c_uint),
+        ("source", sourcetype),
+        ("dry_run", run_type),
+        ("len", ctypes.c_size_t),
+        ("info", ctypes.c_char * 512),
+        ("software_set", ctypes.c_char * 256),
+        ("running_mode", ctypes.c_char * 256),
+        ("disable_store_swu", ctypes.c_bool)
+    ]
+
+
+class status(ctypes.Structure):
+    _fields_ = [
+        ("current", ctypes.c_int),
+        ("last_result", ctypes.c_int),
+        ("error", ctypes.c_int),
+        ("desc", ctypes.c_char * 2048)
+    ]
+
+
+class notify(ctypes.Structure):
+    _fields_ = [
+        ("status", ctypes.c_int),
+        ("error", ctypes.c_int),
+        ("level", ctypes.c_int),
+        ("msg", ctypes.c_char * 2048)
+    ]
+
+
+class instmsg(ctypes.Structure):
+    _fields_ = [
+        ("req", swupdate_request),
+        ("len", ctypes.c_uint),
+        ("buf", ctypes.c_char * 2048)
+    ]
+
+
+class procmsg(ctypes.Structure):
+    _fields_ = [
+        ("source", sourcetype),
+        ("cmd", ctypes.c_int),
+        ("timeout", ctypes.c_int),
+        ("len", ctypes.c_uint),
+        ("buf", ctypes.c_char * 2048)
+    ]
+
+
+class aeskeymsg(ctypes.Structure):
+    _fields_ = [
+        ("key_ascii", ctypes.c_char * 65),
+        ("ivt_ascii", ctypes.c_char * 33)
+    ]
+
+
+class versions(ctypes.Structure):
+    _fields_ = [
+        ("minimum_version", ctypes.c_char * 256),
+        ("maximum_version", ctypes.c_char * 256),
+        ("current_version", ctypes.c_char * 256),
+        ("update_type", ctypes.c_char * 256)
+    ]
+
+
+class revisions(ctypes.Structure):
+    _fields_ = [
+        ("boardname", ctypes.c_char * 256),
+        ("revision", ctypes.c_char * 256)
+    ]
+
+
+class vars(ctypes.Structure):
+    _fields_ = [
+        ("varnamespace", ctypes.c_char * 256),
+        ("varname", ctypes.c_char * 256),
+        ("varvalue", ctypes.c_char * 256)
+    ]
+
+
+class msgdata(ctypes.Union):
+    _fields_ = [
+        ("msg", ctypes.c_char * 128),
+        ("status", status),
+        ("notify", notify),
+        ("instmsg", instmsg),
+        ("procmsg", procmsg),
+        ("aeskeymsg", aeskeymsg),
+        ("versions", versions),
+        ("revisions", revisions),
+        ("vars", vars)
+    ]
+
+
+class ipc_message(ctypes.Structure):
+    _fields_ = [
+        ("magic", ctypes.c_int),
+        ("type", msgtype),
+        ("data", msgdata)
+    ]

mtda/storage/swupdate.py

@@ -0,0 +1,73 @@
+# ---------------------------------------------------------------------------
+# swupdate storage driver for MTDA
+# ---------------------------------------------------------------------------
+#
+# This software is a part of MTDA.
+# Copyright (C) 2025 Siemens
+#
+# ---------------------------------------------------------------------------
+# SPDX-License-Identifier: MIT
+# ---------------------------------------------------------------------------
+
+import mtda.constants as CONSTS
+from mtda.storage.controller import StorageController
+import mtda.storage.helpers.swupdate_ipc as IPC
+import socket
+import ctypes
+
+
+class SWUpdate(StorageController):
+    def __init__(self, mtda):
+        self.mtda = mtda
+        self.writtenBytes = 0
+        self._ipc_socket = None
+
+    def open(self):
+        """ Open the shared storage device for I/O operations"""
+        self.mtda.debug(2, "swupdate open")
+        self.writtenBytes = 0
+
+        self._ipc_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        # TODO: should come from a constant
+        self._ipc_socket.connect("/var/run/swupdate/sockinstctrl")
+        self._perform_handshake()
+        return True
+
+    def close(self):
+        self._ipc_socket.close()
+        return True
+
+    def status(self):
+        return CONSTS.STORAGE.ON_HOST
+
+    def tell(self):
+        return self.writtenBytes
+
+    def write(self, data):
+        self._ipc_socket.sendall(data)
+        self.writtenBytes += len(data)
+        self.mtda.notify_write()
+        return len(data)
+
+    def _perform_handshake(self):
+        sock = self._ipc_socket
+        sock.sendall(self._create_ipc_header_msg())
+        response = sock.recv(ctypes.sizeof(IPC.ipc_message))
+        ack = IPC.ipc_message.from_buffer_copy(response)
+        if ack.type.value != IPC.msgtype.ACK:
+            raise Exception("SWupdate error")
+
+    def _create_ipc_header_msg(self):
+        # TODO: for testing we create a dryrun message
+        req = IPC.swupdate_request(
+            apiversion=IPC.SWUPDATE_API_VERSION,
+            disable_store_swu=True,
+            source=IPC.sourcetype.SOURCE_NETWORK,
+            dry_run=IPC.run_type.RUN_DRYRUN)
+        instmsg = IPC.instmsg(req=req)
+        msgdata = IPC.msgdata(instmsg=instmsg)
+
+        return IPC.ipc_message(
+            magic=IPC.IPC_MAGIC,
+            type=IPC.msgtype.REQ_INSTALL,
+            data=msgdata)