tcl_ota_check/tclcheck.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# pylint: disable=C0111,C0326

import base64
import hashlib
import random
import sys
import time
import xml.dom.minidom
from collections import OrderedDict
from math import floor
try:
    from defusedxml import ElementTree
except (ImportError, AttributeError):
    from xml.etree import ElementTree
import requests

class FotaCheck:
    VDKEY = "1271941121281905392291845155542171963889169361242115412511417616616958244916823523421516924614377131161951402261451161002051042011757216713912611682532031591181861081836612643016596231212872211620511861302106446924625728571011411121471811641125920123641181975581511602312222261817375462445966911723844130106116313122624220514"

    CKTP_CHECKAUTO = 1
    CKTP_CHECKMANUAL = 2
    MODE_OTA = 2
    MODE_FULL = 4
    RTD_ROOTED = 2
    RTD_UNROOTED = 1
    CHNL_WIFI = 2
    CHNL_3G = 1

    def __init__(self):
        self.serid = "543212345000000"
        self.curef = "PRD-63117-011"
        self.fv    = "AAM481"
        self.osvs  = "7.1.1"
        self.mode  = self.MODE_FULL
        self.ftype = "Firmware"
        self.cltp  = 10
        self.cktp  = self.CKTP_CHECKMANUAL
        self.rtd   = self.RTD_UNROOTED
        self.chnl  = self.CHNL_WIFI
        self.sess = requests.Session()
        self.g2master = self.get_master_server()
        #self.req.headers.update({"User-Agent": "com.tcl.fota/5.1.0.2.0029.0, Android"})
        self.sess.headers.update({"User-Agent": "tcl"})

    @staticmethod
    def get_salt():
        millis = floor(time.time() * 1000)
        tail = "{:06d}".format(random.randint(0, 999999))
        return "{}{}".format(str(millis), tail)

    @staticmethod
    def get_master_server():
        master_servers = [
            "g2master-us-east.tclclouds.com",
            "g2master-us-west.tclclouds.com",
            "g2master-eu-west.tclclouds.com",
            "g2master-ap-south.tclclouds.com",
            "g2master-ap-north.tclclouds.com",
            "g2master-sa-east.tclclouds.com",
        ]
        return random.choice(master_servers)

    def do_check(self):
        url = "https://" + self.g2master + "/check.php"
        params = OrderedDict()
        params["id"]    = self.serid
        params["curef"] = self.curef
        params["fv"]    = self.fv
        params["mode"]  = self.mode
        params["type"]  = self.ftype
        params["cltp"]  = self.cltp
        params["cktp"]  = self.cktp
        params["rtd"]   = self.rtd
        params["chnl"]  = self.chnl
        #params["osvs"]  = self.osvs

        req = self.sess.get(url, params=params)
        if req.status_code == 200:
            return req.text
        else:
            print(repr(req))
            print(repr(req.headers))
            print(repr(req.text))
            raise SystemExit

    @staticmethod
    def pretty_xml(xmlstr):
        mdx = xml.dom.minidom.parseString(xmlstr)
        return mdx.toprettyxml(indent="  ")

    @staticmethod
    def parse_check(xmlstr):
        root = ElementTree.fromstring(xmlstr)
        curef = root.find("CUREF").text
        fv = root.find("VERSION").find("FV").text
        tv = root.find("VERSION").find("TV").text
        fw_id = root.find("FIRMWARE").find("FW_ID").text
        fileinfo = root.find("FIRMWARE").find("FILESET").find("FILE")
        fileid   = fileinfo.find("FILE_ID").text
        filename = fileinfo.find("FILENAME").text
        filesize = fileinfo.find("SIZE").text
        filehash = fileinfo.find("CHECKSUM").text
        return curef, fv, tv, fw_id, fileid, filename, filesize, filehash

    def get_vk2(self, params_dict):
        params_dict["cltp"] = 10
        query = ""
        for k, v in params_dict.items():
            if len(query) > 0:
                query += "&"
            query += k + "=" + str(v)
        query += self.VDKEY
        engine = hashlib.sha1()
        engine.update(bytes(query, "utf-8"))
        hexhash = engine.hexdigest()
        return hexhash

    '''
        private HashMap<String, String> buildDownloadUrisParams(UpdatePackageInfo updatePackageInfo) {
            FotaLog.m28v(TAG, "doAfterCheck");
            String salt = FotaUtil.salt();
            HashMap linkedHashMap = new LinkedHashMap();
            linkedHashMap.put("id", this.internalBuilder.getParam("id"));
            linkedHashMap.put("salt", salt);
            linkedHashMap.put("curef", updatePackageInfo.mCuref);
            linkedHashMap.put("fv", updatePackageInfo.mFv);
            linkedHashMap.put("tv", updatePackageInfo.mTv);
            linkedHashMap.put("type", "Firmware");
            linkedHashMap.put("fw_id", updatePackageInfo.mFirmwareId);
            linkedHashMap.put("mode", "2");
            linkedHashMap.put("vk", generateVk2((LinkedHashMap) linkedHashMap.clone()));
            linkedHashMap.put("cltp", "10");
            linkedHashMap.put("cktp", this.internalBuilder.getParam("cktp"));
            linkedHashMap.put("rtd", this.internalBuilder.getParam("rtd"));
            linkedHashMap.put("chnl", this.internalBuilder.getParam("chnl"));
            return linkedHashMap;
        }
    '''

    def do_request(self, curef, fv, tv, fw_id):
        url = "https://" + self.g2master + "/download_request.php"
        params = OrderedDict()
        params["id"]    = self.serid
        params["salt"]  = self.get_salt()
        params["curef"] = curef
        params["fv"]    = fv
        params["tv"]    = tv
        params["type"]  = self.ftype
        params["fw_id"] = fw_id
        params["mode"]  = self.mode
        params["vk"]    = self.get_vk2(params)
        params["cltp"]  = self.cltp
        params["cktp"]  = self.cktp
        params["rtd"]   = self.rtd
        if self.mode == self.MODE_FULL:
            params["foot"]  = 1
        params["chnl"]  = self.chnl

        #print(repr(dict(params)))
        req = self.sess.post(url, data=params)
        if req.status_code == 200:
            return req.text
        else:
            print(repr(req))
            print(repr(req.headers))
            print(repr(req.text))
            raise SystemExit

    @staticmethod
    def parse_request(xmlstr):
        root = ElementTree.fromstring(xmlstr)
        file = root.find("FILE_LIST").find("FILE")
        fileid  = file.find("FILE_ID").text
        fileurl = file.find("DOWNLOAD_URL").text
        slave_list = root.find("SLAVE_LIST").findall("SLAVE")
        enc_list = root.find("SLAVE_LIST").findall("ENCRYPT_SLAVE")
        slaves = [s.text for s in slave_list]
        encslaves = [s.text for s in enc_list]
        return fileid, fileurl, slaves, encslaves

    def encrypt_header(self, address, encslave):
        encs = dict()
        encs[b"YWNjb3VudA=="] = b"emhlbmdodWEuZ2Fv"
        encs[b"cGFzc3dvcmQ="] = b"cWFydUQ0b2s="
        params = {base64.b64decode(key): base64.b64decode(val) for key, val in encs.items()}
        params[b"address"] = bytes(address, "utf-8")
        url = "http://" + encslave + "/encrypt_header.php"
        req = self.sess.post(url, data=params)
        if req.status_code == 206:  # partial
            #return req.text
            contentlength = int(req.headers["Content-Length"])
            sentinel = "\nHEADER FOUND" if contentlength == 4194320 else "\nNO HEADER FOUND"
            return sentinel
        else:
            print(repr(req))
            print(repr(req.headers))
            raise SystemExit


if __name__ == "__main__":
    fc = FotaCheck()
    fc.serid = "3531510"
    fc.curef = "PRD-63117-011"
    fc.fv    = "AAM481"
    #fc.osvs  = "7.1.1"
    fc.mode  = fc.MODE_OTA
    fc.cltp  = 10

    check_xml = fc.do_check()
    #print(fc.pretty_xml(check_xml))
    curef, fv, tv, fw_id, fileid, fn, fsize, fhash = fc.parse_check(check_xml)

    req_xml = fc.do_request(curef, fv, tv, fw_id)
    print(fc.pretty_xml(req_xml))
    fileid, fileurl, slaves, encslaves = fc.parse_request(req_xml)

    for s in slaves:
        print("http://{}{}".format(s, fileurl))

    if fc.mode == fc.MODE_FULL:
        print(fc.encrypt_header(fileurl, random.choice(encslaves)))