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 requests
   6 import hashlib
   7 import os
   8 import tools.config
   9 from tools import helpers
  10
  11
  12 def sha256sum(filename):
  13     h = hashlib.sha256()
  14     b = bytearray(128*1024)
  15     mv = memoryview(b)
  16     with open(filename, 'rb', buffering=0) as f:
  17         for n in iter(lambda: f.readinto(mv), 0):
  18             h.update(mv[:n])
  19     return h.hexdigest()
  20
  21
  22 def get(args):
  23     cfg = tools.config.load(args)
  24     system_ota = cfg["waydroid"]["system_ota"]
  25     system_request = requests.get(system_ota)
  26     if system_request.status_code != 200:
  27         raise ValueError(
  28             "Failed to get system OTA channel: {}".format(system_ota))
  29     system_responses = system_request.json()["response"]
  30     if len(system_responses) < 1:
  31         raise ValueError("No images found on system channel")
  32
  33     for system_response in system_responses:
  34         if system_response['datetime'] > int(cfg["waydroid"]["system_datetime"]):
  35             images_zip = helpers.http.download(
  36                 args, system_response['url'], system_response['filename'], cache=False)
  37             logging.info("Validating system image")
  38             if sha256sum(images_zip) != system_response['id']:
  39                 raise ValueError("Downloaded system image hash doesn't match, expected: {}".format(
  40                     system_response['id']))
  41             logging.info("Extracting to " + args.images_path)
  42             with zipfile.ZipFile(images_zip, 'r') as zip_ref:
  43                 zip_ref.extractall(args.images_path)
  44             cfg["waydroid"]["system_datetime"] = str(system_response['datetime'])
  45             tools.config.save(args, cfg)
  46             os.remove(images_zip)
  47             break
  48
  49     vendor_ota = cfg["waydroid"]["vendor_ota"]
  50     vendor_request = requests.get(vendor_ota)
  51     if vendor_request.status_code != 200:
  52         raise ValueError(
  53             "Failed to get vendor OTA channel: {}".format(vendor_ota))
  54     vendor_responses = vendor_request.json()["response"]
  55     if len(vendor_responses) < 1:
  56         raise ValueError("No images found on vendor channel")
  57
  58     for vendor_response in vendor_responses:
  59         if vendor_response['datetime'] > int(cfg["waydroid"]["vendor_datetime"]):
  60             images_zip = helpers.http.download(
  61                 args, vendor_response['url'], vendor_response['filename'], cache=False)
  62             logging.info("Validating vendor image")
  63             if sha256sum(images_zip) != vendor_response['id']:
  64                 raise ValueError("Downloaded vendor image hash doesn't match, expected: {}".format(
  65                     vendor_response['id']))
  66             logging.info("Extracting to " + args.images_path)
  67             with zipfile.ZipFile(images_zip, 'r') as zip_ref:
  68                 zip_ref.extractall(args.images_path)
  69             cfg["waydroid"]["vendor_datetime"] = str(vendor_response['datetime'])
  70             tools.config.save(args, cfg)
  71             os.remove(images_zip)
  72             break
  73
  74 def replace(args, system_zip, system_time, vendor_zip, vendor_time):
  75     cfg = tools.config.load(args)
  76     args.images_path = cfg["waydroid"]["images_path"]
  77     if os.path.exists(system_zip):
  78         with zipfile.ZipFile(system_zip, 'r') as zip_ref:
  79             zip_ref.extractall(args.images_path)
  80         cfg["waydroid"]["system_datetime"] = str(system_time)
  81         tools.config.save(args, cfg)
  82     if os.path.exists(vendor_zip):
  83         with zipfile.ZipFile(vendor_zip, 'r') as zip_ref:
  84             zip_ref.extractall(args.images_path)
  85         cfg["waydroid"]["vendor_datetime"] = str(vendor_time)
  86         tools.config.save(args, cfg)
  87
  88
  89 def mount_rootfs(args, images_dir):
  90     helpers.mount.mount(args, images_dir + "/system.img",
  91                         tools.config.defaults["rootfs"], umount=True)
  92     helpers.mount.mount(args, images_dir + "/vendor.img",
  93                            tools.config.defaults["rootfs"] + "/vendor")
  94     for egl_path in ["/vendor/lib/egl", "/vendor/lib64/egl"]:
  95         if os.path.isdir(egl_path):
  96             helpers.mount.bind(
  97                 args, egl_path, tools.config.defaults["rootfs"] + egl_path)
  98     if helpers.mount.ismount("/odm"):
  99         helpers.mount.bind(
 100             args, "/odm", tools.config.defaults["rootfs"] + "/odm_extra")
 101     else:
 102         if os.path.isdir("/vendor/odm"):
 103             helpers.mount.bind(
 104                 args, "/vendor/odm", tools.config.defaults["rootfs"] + "/odm_extra")
 105     helpers.mount.bind_file(args, args.work + "/waydroid.prop",
 106                             tools.config.defaults["rootfs"] + "/vendor/waydroid.prop")
 107
 108 def umount_rootfs(args):
 109     helpers.mount.umount_all(args, tools.config.defaults["rootfs"])