hardware_manager: Validate upgrade zips against the saved ota channel

[waydroid.git] / tools / services / hardware_manager.py

diff --git a/tools/services/hardware_manager.py b/tools/services/hardware_manager.py
index 529348dc102f5e521444842d1d08af6f47f36b7b..b9cff466879779f1ddbd41bc71ca0264e9e2d768 100644 (file)
--- a/tools/services/hardware_manager.py
+++ b/tools/services/hardware_manager.py
@@ -2,10 +2,14 @@
 # SPDX-License-Identifier: GPL-3.0-or-later
 import logging
 import threading
+import os
 import tools.actions.container_manager
+import tools.actions.session_manager
+import tools.config
 from tools import helpers
 from tools.interfaces import IHardware
 
+stopping = False
 
 def start(args):
     def enableNFC(enable):
@@ -15,28 +19,45 @@ def start(args):
         logging.debug("Function enableBluetooth not implemented")
 
     def suspend():
-        tools.actions.container_manager.freeze(args)
+        cfg = tools.config.load(args)
+        if cfg["waydroid"]["suspend_action"] == "stop":
+            tools.actions.session_manager.stop(args)
+        else:
+            tools.actions.container_manager.freeze(args)
 
     def reboot():
         helpers.lxc.stop(args)
         helpers.lxc.start(args)
 
     def upgrade(system_zip, system_time, vendor_zip, vendor_time):
+        if os.path.exists(system_zip) and not helpers.images.validate(args, "system_ota", system_zip):
+            logging.warning("Not upgrading because system.img comes from an unverified source")
+            return
+        if os.path.exists(vendor_zip) and not helpers.images.validate(args, "vendor_ota", vendor_zip):
+            logging.warning("Not upgrading because vendor.img comes from an unverified source")
+            return
         helpers.lxc.stop(args)
         helpers.images.umount_rootfs(args)
         helpers.images.replace(args, system_zip, system_time,
                                vendor_zip, vendor_time)
-        helpers.images.mount_rootfs(args, args.images_path)
+        args.session["background_start"] = "false"
+        helpers.images.mount_rootfs(args, args.images_path, args.session)
+        helpers.protocol.set_aidl_version(args)
         helpers.lxc.start(args)
 
     def service_thread():
-        IHardware.add_service(
-            args, enableNFC, enableBluetooth, suspend, reboot, upgrade)
+        while not stopping:
+            IHardware.add_service(
+                args, enableNFC, enableBluetooth, suspend, reboot, upgrade)
 
+    global stopping
+    stopping = False
     args.hardware_manager = threading.Thread(target=service_thread)
     args.hardware_manager.start()
 
 def stop(args):
+    global stopping
+    stopping = True
     try:
         if args.hardwareLoop:
             args.hardwareLoop.quit()