Samsung firmware downloader (warning! code quality isn't very good!)

samfirm.py

#!/usr/bin/python -u
# Based on:
# https://github.com/DavidArsene/samfirm.js
# https://github.com/martinetd/samloader
# https://github.com/ysfchn/easyfirmware
from Cryptodome.Cipher import AES
from Cryptodome.Util.Padding import pad
from Cryptodome.Util.Padding import unpad
import base64
import hashlib
import os
import requests
import sys
import xml.etree.ElementTree as ET
try:
    from stream_unzip import stream_unzip
except ImportError:
    stream_unzip = None
    pass

def info(text): print(f"\033[94mINFO: \033[00m{text}")
def warning(text): print(f"\033[93mWARNING: \033[00m{text}")
def error(text):
    print(f"\033[91mERROR: \033[00m{text}")
    quit()

def getlatest(model: str, csc: str):
    url = f"https://fota-cloud-dn.ospserver.net/firmware/{csc}/{model}/version.xml"
    try:
        response = requests.get(url, headers={"User-Agent": "Kies2.0_FUS"}).text
        tree = ET.fromstring(response)
        for version in tree.iter("latest"):
            if version.text == None:
                error(f"No updates found for {model}, {csc}")
            fwver = version.text.split("/")
            return fwver
        else:
            raise Exception
    except:
        error(f"Encountered failure while attempting to get latest version for {model}, {csc}")

def decryptnonce(encrypted_nonce: str):
    NONCE_KEY = "vicopx7dqu06emacgpnpy8j8zwhduwlh"
    AUTH_KEY = "9u7qab84rpc16gvk"
    cipher = AES.new(
        key = NONCE_KEY.encode(),
        mode = AES.MODE_CBC,
        iv = NONCE_KEY.encode()[:16]
    )
    decrypted_nonce = unpad(cipher.decrypt(base64.b64decode(encrypted_nonce)), AES.block_size)
    sigkey = "".join(map(lambda i: NONCE_KEY[decrypted_nonce[i] % 16], range(16))) + AUTH_KEY
    cipher = AES.new(
        key = sigkey.encode(),
        mode = AES.MODE_CBC,
        iv = sigkey.encode()[:16]
    )
    signature = cipher.encrypt(pad(decrypted_nonce, AES.block_size))
    return decrypted_nonce.decode(), base64.b64encode(signature).decode()

def logiccheck(input, checkvalue):
    return "".join([input[ord(x) & 0xF] for x in checkvalue])

if len(sys.argv) != 4:
    info("Usage: samfirm.py <model> <region> <imei/serial number>")

model = sys.argv[1].upper()
region = sys.argv[2].upper()
imei = sys.argv[3].upper()

if len(region) != 3:
    error("Invalid region code")

if len(imei) < 11:
    error("Invalid IMEI or serial number")

pda, csc, modem = getlatest(model, region)
version = f"{pda}/{csc}/{modem if modem != '' else pda}/{pda}"
print(f"Latest version:\n  PDA: {pda}\n  CSC: {csc}\n  MODEM: {modem if modem != '' else 'N/A'}")
session = requests.post("https://neofussvr.sslcs.cdngc.net/NF_DownloadGenerateNonce.do", headers={"Authorization": 'FUS nonce="", signature="", nc="", type="", realm="", newauth="1"', "User-Agent": "Kies2.0_FUS", "Accept": "application/xml"})
encrypted_nonce = session.headers["NONCE"]
decrypted_nonce, signature = decryptnonce(encrypted_nonce)

cookie = session.headers["Set-Cookie"].split("; ")
for i in cookie:
    if "JSESSIONID=" in i:
        jsessionid = i.split("JSESSIONID=", 1)[1]
binary_info_xml = f"<FUSMsg><FUSHdr><ProtoVer>1.0</ProtoVer></FUSHdr><FUSBody><Put><ACCESS_MODE><Data>2</Data></ACCESS_MODE><BINARY_NATURE><Data>1</Data></BINARY_NATURE><CLIENT_PRODUCT><Data>Smart Switch</Data></CLIENT_PRODUCT><CLIENT_VERSION><Data>4.3.23123_1</Data></CLIENT_VERSION><DEVICE_FW_VERSION><Data>{version}</Data></DEVICE_FW_VERSION><DEVICE_LOCAL_CODE><Data>{region}</Data></DEVICE_LOCAL_CODE><DEVICE_MODEL_NAME><Data>{model}</Data></DEVICE_MODEL_NAME><DEVICE_IMEI_PUSH><Data>{imei}</Data></DEVICE_IMEI_PUSH><LOGIC_CHECK><Data>{logiccheck(version, decrypted_nonce)}</Data></LOGIC_CHECK>{'<DEVICE_AID_CODE><Data>EUX</Data></DEVICE_AID_CODE><DEVICE_CC_CODE><Data>DE</Data></DEVICE_CC_CODE><MCC_NUM><Data>262</Data></MCC_NUM><MNC_NUM><Data>01</Data></MNC_NUM>' if region == 'EUX' else '<DEVICE_AID_CODE><Data>EUY</Data></DEVICE_AID_CODE><DEVICE_CC_CODE><Data>RS</Data></DEVICE_CC_CODE><MCC_NUM><Data>220</Data></MCC_NUM><MNC_NUM><Data>01</Data></MNC_NUM>' if region == 'EUY' else ''}</Put></FUSBody></FUSMsg>"
binary_info_headers = {
    "User-Agent": "Kies2.0_FUS",
    "Authorization": f'FUS nonce="{str(encrypted_nonce)}", signature="{signature}", nc="", type="", realm="", newauth="1"',
    "Cookie": f"JSESSIONID={jsessionid}",
    "Accept": "application/xml",
    "Content-Type": "application/xml"
}
binary_info = requests.post("https://neofussvr.sslcs.cdngc.net/NF_DownloadBinaryInform.do", data=binary_info_xml, headers=binary_info_headers)
if "<Status>200</Status>" not in binary_info.text:
    error("FUS Server received a bad request. Please check your input")

encrypted_nonce = binary_info.headers["NONCE"]
decrypted_nonce, signature = decryptnonce(encrypted_nonce)

tree = ET.fromstring(binary_info.text)
filename = tree.find("./FUSBody/Put/BINARY_NAME/Data").text
model_path = tree.find("./FUSBody/Put/MODEL_PATH/Data").text
logic_value = tree.find("./FUSBody/Put/LOGIC_VALUE_FACTORY/Data").text

filename_logic = filename.split(".")[0][-16:]

decryption_key = hashlib.md5(logiccheck(version, logic_value).encode()).digest()
cipher = AES.new(decryption_key, AES.MODE_ECB)

binary_init_xml = f"<FUSMsg><FUSHdr><ProtoVer>1.0</ProtoVer></FUSHdr><FUSBody><Put><BINARY_FILE_NAME><Data>{filename}</Data></BINARY_FILE_NAME><LOGIC_CHECK><Data>{logiccheck(filename_logic, decrypted_nonce)}</Data></LOGIC_CHECK></Put></FUSBody></FUSMsg>"
binary_init_headers = {
    "User-Agent": "Kies2.0_FUS",
    "Authorization": f'FUS nonce="{encrypted_nonce}", signature="{signature}", nc="", type="", realm="", newauth="1"',
    "Cookie": f"JSESSIONID={jsessionid}",
    "Accept": "application/xml",
    "Content-Type": "application/xml"
}
binary_init = requests.post("https://neofussvr.sslcs.cdngc.net/NF_DownloadBinaryInitForMass.do", data=binary_init_xml, headers=binary_init_headers)
encrypted_nonce=binary_init.headers["NONCE"]
decrypted_nonce, signature = decryptnonce(encrypted_nonce)

binary_download_headers = {
    "User-Agent": "Kies2.0_FUS",
    "Authorization": f'FUS nonce="{encrypted_nonce}", signature="{signature}", nc="", type="", realm="", newauth="1"',
    "Cookie": f"JSESSIONID={jsessionid}",
    "Accept": "application/xml",
    "Content-Type": "application/xml"
}

def download():
    with requests.get(f"http://cloud-neofussvr.samsungmobile.com/NF_DownloadBinaryForMass.do?file={model_path}{filename}", headers=binary_download_headers, stream=True) as req:
        req.raise_for_status()
        for chunk in req.iter_content(chunk_size=65536):
            if chunk:
                decrypted = cipher.decrypt(chunk)
                yield decrypted

if stream_unzip != None:
    os.makedirs(model, exist_ok=True)
    for file_name, file_size, unzipped_chunks in stream_unzip(download()):
        with open(f"{model}/{file_name.decode()}", 'wb') as f:
            for data in unzipped_chunks:
                f.write(data)
else:
    with open(f"{model}_{region}_{pda}_fac.zip", 'wb') as f:
        for data in download():
            f.write(data)