tools/helpers/images.py

   1 # Copyright 2021 Erfan Abdi
   2 # SPDX-License-Identifier: GPL-3.0-or-later
   3 import logging
   4 import zipfile
   5 import json
   6 import hashlib
   7 import shutil
   8 import os
   9 import tools.config
  10 from tools import helpers
  11 from shutil import which
  12
  13 def sha256sum(filename):
  14     h = hashlib.sha256()
  15     b = bytearray(128*1024)
  16     mv = memoryview(b)
  17     with open(filename, 'rb', buffering=0) as f:
  18         for n in iter(lambda: f.readinto(mv), 0):
  19             h.update(mv[:n])
  20     return h.hexdigest()
  21
  22
  23 def get(args):
  24     cfg = tools.config.load(args)
  25     system_ota = cfg["waydroid"]["system_ota"]
  26     system_request = helpers.http.retrieve(system_ota)
  27     if system_request[0] != 200:
  28         raise ValueError(
  29             "Failed to get system OTA channel: {}, error: {}".format(args.system_ota, system_request[0]))
  30     system_responses = json.loads(system_request[1].decode('utf8'))["response"]
  31     if len(system_responses) < 1:
  32         raise ValueError("No images found on system channel")
  33
  34     for system_response in system_responses:
  35         if system_response['datetime'] > int(cfg["waydroid"]["system_datetime"]):
  36             images_zip = helpers.http.download(
  37                 args, system_response['url'], system_response['filename'], cache=False)
  38             logging.info("Validating system image")
  39             if sha256sum(images_zip) != system_response['id']:
  40                 try:
  41                     os.remove(images_zip)
  42                 except:
  43                     pass
  44                 raise ValueError("Downloaded system image hash doesn't match, expected: {}".format(
  45                     system_response['id']))
  46             logging.info("Extracting to " + args.images_path)
  47             with zipfile.ZipFile(images_zip, 'r') as zip_ref:
  48                 zip_ref.extractall(args.images_path)
  49             cfg["waydroid"]["system_datetime"] = str(system_response['datetime'])
  50             tools.config.save(args, cfg)
  51             os.remove(images_zip)
  52             break
  53
  54     vendor_ota = cfg["waydroid"]["vendor_ota"]
  55     vendor_request = helpers.http.retrieve(vendor_ota)
  56     if vendor_request[0] != 200:
  57         raise ValueError(
  58             "Failed to get vendor OTA channel: {}, error: {}".format(vendor_ota, vendor_request[0]))
  59     vendor_responses = json.loads(vendor_request[1].decode('utf8'))["response"]
  60     if len(vendor_responses) < 1:
  61         raise ValueError("No images found on vendor channel")
  62
  63     for vendor_response in vendor_responses:
  64         if vendor_response['datetime'] > int(cfg["waydroid"]["vendor_datetime"]):
  65             images_zip = helpers.http.download(
  66                 args, vendor_response['url'], vendor_response['filename'], cache=False)
  67             logging.info("Validating vendor image")
  68             if sha256sum(images_zip) != vendor_response['id']:
  69                 try:
  70                     os.remove(images_zip)
  71                 except:
  72                     pass
  73                 raise ValueError("Downloaded vendor image hash doesn't match, expected: {}".format(
  74                     vendor_response['id']))
  75             logging.info("Extracting to " + args.images_path)
  76             with zipfile.ZipFile(images_zip, 'r') as zip_ref:
  77                 zip_ref.extractall(args.images_path)
  78             cfg["waydroid"]["vendor_datetime"] = str(vendor_response['datetime'])
  79             tools.config.save(args, cfg)
  80             os.remove(images_zip)
  81             break
  82     remove_overlay(args)
  83
  84 def validate(args, channel, image_zip):
  85     # Verify that the zip comes from the channel
  86     cfg = tools.config.load(args)
  87     channel_url = cfg["waydroid"][channel]
  88     channel_request = helpers.http.retrieve(channel_url)
  89     if channel_request[0] != 200:
  90         return False
  91     channel_responses = json.loads(channel_request[1].decode('utf8'))["response"]
  92     for build in channel_responses:
  93         if sha256sum(image_zip) == build['id']:
  94             return True
  95     logging.warning(f"Could not verify the image {image_zip} against {channel_url}")
  96     return False
  97
  98 def replace(args, system_zip, system_time, vendor_zip, vendor_time):
  99     cfg = tools.config.load(args)
 100     args.images_path = cfg["waydroid"]["images_path"]
 101     if os.path.exists(system_zip):
 102         with zipfile.ZipFile(system_zip, 'r') as zip_ref:
 103             zip_ref.extractall(args.images_path)
 104         os.remove(system_zip)
 105         cfg["waydroid"]["system_datetime"] = str(system_time)
 106         tools.config.save(args, cfg)
 107     if os.path.exists(vendor_zip):
 108         with zipfile.ZipFile(vendor_zip, 'r') as zip_ref:
 109             zip_ref.extractall(args.images_path)
 110         os.remove(vendor_zip)
 111         cfg["waydroid"]["vendor_datetime"] = str(vendor_time)
 112         tools.config.save(args, cfg)
 113     remove_overlay(args)
 114
 115 def remove_overlay(args):
 116     if os.path.isdir(tools.config.defaults["overlay_rw"]):
 117         shutil.rmtree(tools.config.defaults["overlay_rw"])
 118     if os.path.isdir(tools.config.defaults["overlay_work"]):
 119         shutil.rmtree(tools.config.defaults["overlay_work"])
 120
 121 def make_prop(args, cfg, full_props_path):
 122     if not os.path.isfile(args.work + "/waydroid_base.prop"):
 123         raise RuntimeError("waydroid_base.prop Not found")
 124     with open(args.work + "/waydroid_base.prop") as f:
 125         props = f.read().splitlines()
 126     if not props:
 127         raise RuntimeError("waydroid_base.prop is broken!!?")
 128
 129     def add_prop(key, cfg_key):
 130         value = cfg[cfg_key]
 131         if value != "None":
 132             value = value.replace("/mnt/", "/mnt_extra/")
 133             props.append(key + "=" + value)
 134
 135     add_prop("waydroid.host.user", "user_name")
 136     add_prop("waydroid.host.uid", "user_id")
 137     add_prop("waydroid.host.gid", "group_id")
 138     add_prop("waydroid.host_data_path", "waydroid_data")
 139     add_prop("waydroid.xdg_runtime_dir", "xdg_runtime_dir")
 140     add_prop("waydroid.pulse_runtime_path", "pulse_runtime_path")
 141     add_prop("waydroid.wayland_display", "wayland_display")
 142     add_prop("waydroid.background_start", "background_start")
 143     if which("waydroid-sensord") is None:
 144         props.append("waydroid.stub_sensors_hal=1")
 145     dpi = cfg["lcd_density"]
 146     if dpi != "0":
 147         props.append("ro.sf.lcd_density=" + dpi)
 148
 149     final_props = open(full_props_path, "w")
 150     for prop in props:
 151         final_props.write(prop + "\n")
 152     final_props.close()
 153     os.chmod(full_props_path, 0o644)
 154
 155 def mount_rootfs(args, images_dir, session):
 156     cfg = tools.config.load(args)
 157     helpers.mount.mount(args, images_dir + "/system.img",
 158                         tools.config.defaults["rootfs"], umount=True)
 159     if cfg["waydroid"]["mount_overlays"] == "True":
 160         try:
 161             helpers.mount.mount_overlay(args, [tools.config.defaults["overlay"],
 162                                                tools.config.defaults["rootfs"]],
 163                                     tools.config.defaults["rootfs"],
 164                                     upper_dir=tools.config.defaults["overlay_rw"] + "/system",
 165                                     work_dir=tools.config.defaults["overlay_work"] + "/system")
 166         except RuntimeError:
 167             cfg["waydroid"]["mount_overlays"] = "False"
 168             tools.config.save(args, cfg)
 169             logging.warning("Mounting overlays failed. The feature has been disabled.")
 170
 171     helpers.mount.mount(args, images_dir + "/vendor.img",
 172                            tools.config.defaults["rootfs"] + "/vendor")
 173     if cfg["waydroid"]["mount_overlays"] == "True":
 174         helpers.mount.mount_overlay(args, [tools.config.defaults["overlay"] + "/vendor",
 175                                            tools.config.defaults["rootfs"] + "/vendor"],
 176                                     tools.config.defaults["rootfs"] + "/vendor",
 177                                     upper_dir=tools.config.defaults["overlay_rw"] + "/vendor",
 178                                     work_dir=tools.config.defaults["overlay_work"] + "/vendor")
 179
 180     for egl_path in ["/vendor/lib/egl", "/vendor/lib64/egl"]:
 181         if os.path.isdir(egl_path):
 182             helpers.mount.bind(
 183                 args, egl_path, tools.config.defaults["rootfs"] + egl_path)
 184     if helpers.mount.ismount("/odm"):
 185         helpers.mount.bind(
 186             args, "/odm", tools.config.defaults["rootfs"] + "/odm_extra")
 187     else:
 188         if os.path.isdir("/vendor/odm"):
 189             helpers.mount.bind(
 190                 args, "/vendor/odm", tools.config.defaults["rootfs"] + "/odm_extra")
 191
 192     make_prop(args, session, args.work + "/waydroid.prop")
 193     helpers.mount.bind_file(args, args.work + "/waydroid.prop",
 194                             tools.config.defaults["rootfs"] + "/vendor/waydroid.prop")
 195
 196 def umount_rootfs(args):
 197     helpers.mount.umount_all(args, tools.config.defaults["rootfs"])