make_priv_skeleton() - Code Metrics - Inspection of "privlookup->droidlookup, DTEK50 support" - thurask/bbarchivist - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 473d55...344b16 )

by John

created 2016-08-09 22:14 UTC

make_priv_skeleton() A

↳ Parent: make_droid_skeleton()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
dl	0
loc	19
rs	9.4285
c	0
b	0
f	0

#!/usr/bin/env python3
"""This module is used for network connections; APIs, downloading, etc."""

import os  # filesystem read
import xml.etree.ElementTree  # XML parsing
import re  # regexes
import hashlib  # base url creation
import concurrent.futures  # multiprocessing/threading
import glob  # pem file lookup
import requests  # downloading
from bs4 import BeautifulSoup  # scraping
from bbarchivist import utilities  # parse filesize
from bbarchivist.bbconstants import SERVERS  # lookup servers

__author__ = "Thurask"
__license__ = "WTFPL v2"
__copyright__ = "Copyright 2015-2016 Thurask"


def grab_pem():
    """
    Work with either local cacerts or system cacerts. Since cx_freeze is dumb.
    """
    try:
        pemfile = glob.glob(os.path.join(os.getcwd(), "cacert.pem"))[0]
    except IndexError:
        return requests.certs.where()  # no local cacerts
    else:
        return os.path.abspath(pemfile)  # local cacerts


def get_length(url):
    """
    Get content-length header from some URL.

    :param url: The URL to check.
    :type url: str
    """
    os.environ["REQUESTS_CA_BUNDLE"] = grab_pem()
    if url is None:
        return 0
    try:
        heads = requests.head(url)
        fsize = heads.headers['content-length']
        return int(fsize)
    except requests.ConnectionError:
        return 0


def download(url, output_directory=None):
    """
    Download file from given URL.

    :param url: URL to download from.
    :type url: str

    :param output_directory: Download folder. Default is local.
    :type output_directory: str
    """
    if output_directory is None:
        output_directory = os.getcwd()
    lfname = url.split('/')[-1]
    sname = utilities.stripper(lfname)
    os.environ["REQUESTS_CA_BUNDLE"] = grab_pem()
    fname = os.path.join(output_directory, lfname)
    with open(fname, "wb") as file:
        req = requests.get(url, stream=True)
        clength = req.headers['content-length']
        fsize = utilities.fsizer(clength)
        if req.status_code == 200:  # 200 OK
            print("DOWNLOADING {0} [{1}]".format(sname, fsize))
            for chunk in req.iter_content(chunk_size=1024):
                file.write(chunk)
        else:
            print("ERROR: HTTP {0} IN {1}".format(req.status_code, lfname))


def download_bootstrap(urls, outdir=None, workers=5):
    """
    Run downloaders for each file in given URL iterable.

    :param urls: URLs to download.
    :type urls: list

    :param outdir: Download folder. Default is handled in :func:`download`.
    :type outdir: str

    :param workers: Number of worker processes. Default is 5.
    :type workers: int
    """
    workers = len(urls) if len(urls) < workers else workers
    spinman = utilities.SpinManager()
    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as xec:
        try:
            spinman.start()
            for url in urls:
                xec.submit(download, url, outdir)
        except (KeyboardInterrupt, SystemExit):  # pragma: no cover
            xec.shutdown()
            spinman.stop()
    spinman.stop()
    utilities.spinner_clear()
    utilities.line_begin()


def create_base_url(softwareversion):
    """
    Make the root URL for production server files.

    :param softwareversion: Software version to hash.
    :type softwareversion: str
    """
    # Hash software version
    swhash = hashlib.sha1(softwareversion.encode('utf-8'))
    hashedsoftwareversion = swhash.hexdigest()
    # Root of all urls
    baseurl = "http://cdn.fs.sl.blackberry.com/fs/qnx/production/" + hashedsoftwareversion
    return baseurl


def availability(url):
    """
    Check HTTP status code of given URL.
    200 or 301-308 is OK, else is not.

    :param url: URL to check.
    :type url: str
    """
    os.environ["REQUESTS_CA_BUNDLE"] = grab_pem()
    try:
        avlty = requests.head(url)
        status = int(avlty.status_code)
        return status == 200 or 300 < status <= 308
    except requests.ConnectionError:
        return False


def clean_availability(results, server):
    """
    Clean availability for autolookup script.

    :param results: Result dict.
    :type results: dict(str: str)

    :param server: Server, key for result dict.
    :type server: str
    """
    marker = "PD" if server == "p" else server.upper()
    rel = results[server.lower()]
    avail = marker if rel != "SR not in system" and rel is not None else "  "
    return rel, avail


def carrier_checker(mcc, mnc):
    """
    Query BlackBerry World to map a MCC and a MNC to a country and carrier.

    :param mcc: Country code.
    :type mcc: int

    :param mnc: Network code.
    :type mnc: int
    """
    url = "http://appworld.blackberry.com/ClientAPI/checkcarrier?homemcc={0}&homemnc={1}&devicevendorid=-1&pin=0".format(
        mcc, mnc)
    user_agent = {'User-agent': 'AppWorld/5.1.0.60'}
    os.environ["REQUESTS_CA_BUNDLE"] = grab_pem()
    req = requests.get(url, headers=user_agent)
    root = xml.etree.ElementTree.fromstring(req.text)
    for child in root:
        if child.tag == "country":
            country = child.get("name")
        if child.tag == "carrier":
            carrier = child.get("name")
    return country, carrier


def return_npc(mcc, mnc):
    """
    Format MCC and MNC into a NPC.

    :param mcc: Country code.
    :type mcc: int

    :param mnc: Network code.
    :type mnc: int
    """
    return "{0}{1}30".format(str(mcc).zfill(3), str(mnc).zfill(3))


def carrier_query(npc, device, upgrade=False, blitz=False, forced=None):
    """
    Query BlackBerry servers, check which update is out for a carrier.

    :param npc: MCC + MNC (see `func:return_npc`)
    :type npc: int

    :param device: Hexadecimal hardware ID.
    :type device: str

    :param upgrade: Whether to use upgrade files. False by default.
    :type upgrade: bool

    :param blitz: Whether or not to create a blitz package. False by default.
    :type blitz: bool

    :param forced: Force a software release.
    :type forced: str
    """
    upg = "upgrade" if upgrade else "repair"
    forced = "latest" if forced is None else forced
    url = "https://cs.sl.blackberry.com/cse/updateDetails/2.2/"
    query = '<?xml version="1.0" encoding="UTF-8"?>'
    query += '<updateDetailRequest version="2.2.1" authEchoTS="1366644680359">'
    query += "<clientProperties>"
    query += "<hardware>"
    query += "<pin>0x2FFFFFB3</pin><bsn>1128121361</bsn>"
    query += "<imei>004401139269240</imei>"
    query += "<id>0x{0}</id>".format(device)
    query += "</hardware>"
    query += "<network>"
    query += "<homeNPC>0x{0}</homeNPC>".format(npc)
    query += "<iccid>89014104255505565333</iccid>"
    query += "</network>"
    query += "<software>"
    query += "<currentLocale>en_US</currentLocale>"
    query += "<legalLocale>en_US</legalLocale>"
    query += "</software>"
    query += "</clientProperties>"
    query += "<updateDirectives>"
    query += '<allowPatching type="REDBEND">true</allowPatching>'
    query += "<upgradeMode>{0}</upgradeMode>".format(upg)
    query += "<provideDescriptions>false</provideDescriptions>"
    query += "<provideFiles>true</provideFiles>"
    query += "<queryType>NOTIFICATION_CHECK</queryType>"
    query += "</updateDirectives>"
    query += "<pollType>manual</pollType>"
    query += "<resultPackageSetCriteria>"
    query += '<softwareRelease softwareReleaseVersion="{0}" />'.format(forced)
    query += "<releaseIndependent>"
    query += '<packageType operation="include">application</packageType>'
    query += "</releaseIndependent>"
    query += "</resultPackageSetCriteria>"
    query += "</updateDetailRequest>"
    header = {"Content-Type": "text/xml;charset=UTF-8"}
    os.environ["REQUESTS_CA_BUNDLE"] = grab_pem()
    req = requests.post(url, headers=header, data=query)
    return parse_carrier_xml(req.text, blitz)


def parse_carrier_xml(data, blitz=False):
    """
    Parse the response to a carrier update request and return the juicy bits.

    :param data: The data to parse.
    :type data: xml

    :param blitz: Whether or not to create a blitz package. False by default.
    :type blitz: bool
    """
    root = xml.etree.ElementTree.fromstring(data)
    sw_exists = root.find('./data/content/softwareReleaseMetadata')
    swver = "N/A" if sw_exists is None else ""
    if sw_exists is not None:
        for child in root.iter("softwareReleaseMetadata"):
            swver = child.get("softwareReleaseVersion")
    files = []
    osver = ""
    radver = ""
    package_exists = root.find('./data/content/fileSets/fileSet')
    if package_exists is not None:
        baseurl = "{0}/".format(package_exists.get("url"))
        for child in root.iter("package"):
            if not blitz:
                files.append(baseurl + child.get("path"))
            else:
                if child.get("type") not in ["system:radio", "system:desktop", "system:os"]:
                    files.append(baseurl + child.get("path"))
            if child.get("type") == "system:radio":
                radver = child.get("version")
            if child.get("type") == "system:desktop":
                osver = child.get("version")
            if child.get("type") == "system:os":
                osver = child.get("version")
            else:
                pass
    return(swver, osver, radver, files)


def sr_lookup(osver, server):
    """
    Software release lookup, with choice of server.
    :data:`bbarchivist.bbconstants.SERVERLIST` for server list.

    :param osver: OS version to lookup, 10.x.y.zzzz.
    :type osver: str

    :param server: Server to use.
    :type server: str
    """
    reg = re.compile(r"(\d{1,4}\.)(\d{1,4}\.)(\d{1,4}\.)(\d{1,4})")
    query = '<?xml version="1.0" encoding="UTF-8"?>'
    query += '<srVersionLookupRequest version="2.0.0"'
    query += ' authEchoTS="1366644680359">'
    query += '<clientProperties><hardware>'
    query += '<pin>0x2FFFFFB3</pin><bsn>1140011878</bsn>'
    query += '<imei>004402242176786</imei><id>0x8D00240A</id>'
    query += '<isBootROMSecure>true</isBootROMSecure>'
    query += '</hardware>'
    query += '<network>'
    query += '<vendorId>0x0</vendorId><homeNPC>0x60</homeNPC>'
    query += '<currentNPC>0x60</currentNPC><ecid>0x1</ecid>'
    query += '</network>'
    query += '<software><currentLocale>en_US</currentLocale>'
    query += '<legalLocale>en_US</legalLocale>'
    query += '<osVersion>{0}</osVersion>'.format(osver)
    query += '<omadmEnabled>false</omadmEnabled>'
    query += '</software></clientProperties>'
    query += '</srVersionLookupRequest>'
    header = {"Content-Type": "text/xml;charset=UTF-8"}
    os.environ["REQUESTS_CA_BUNDLE"] = grab_pem()
    try:
        req = requests.post(server, headers=header, data=query, timeout=1)
    except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
        return "SR not in system"
    try:
        root = xml.etree.ElementTree.fromstring(req.text)
    except xml.etree.ElementTree.ParseError:
        return "SR not in system"
    else:
        packages = root.findall('./data/content/')
        for package in packages:
            if package.text is not None:
                match = reg.match(package.text)
                if match:
                    return package.text
                else:
                    return "SR not in system"


def sr_lookup_bootstrap(osv):
    """
    Run lookups for each server for given OS.

    :param osv: OS to check.
    :type osv: str
    """
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as xec:
        try:
            results = {
                "p": None,
                "a1": None,
                "a2": None,
                "b1": None,
                "b2": None
            }
            for key in results:
                results[key] = xec.submit(sr_lookup, osv, SERVERS[key]).result()
            return results
        except KeyboardInterrupt:  # pragma: no cover
            xec.shutdown(wait=False)


def available_bundle_lookup(mcc, mnc, device):
    """
    Check which software releases were ever released for a carrier.

    :param mcc: Country code.
    :type mcc: int

    :param mnc: Network code.
    :type mnc: int

    :param device: Hexadecimal hardware ID.
    :type device: str
    """
    server = "https://cs.sl.blackberry.com/cse/availableBundles/1.0.0/"
    npc = return_npc(mcc, mnc)
    query = '<?xml version="1.0" encoding="UTF-8"?>'
    query += '<availableBundlesRequest version="1.0.0" '
    query += 'authEchoTS="1366644680359">'
    query += '<deviceId><pin>0x2FFFFFB3</pin></deviceId>'
    query += '<clientProperties><hardware><id>0x{0}</id>'.format(device)
    query += '<isBootROMSecure>true</isBootROMSecure></hardware>'
    query += '<network><vendorId>0x0</vendorId><homeNPC>0x{0}</homeNPC>'.format(npc)
    query += '<currentNPC>0x{0}</currentNPC></network><software>'.format(npc)
    query += '<currentLocale>en_US</currentLocale>'
    query += '<legalLocale>en_US</legalLocale>'
    query += '<osVersion>10.0.0.0</osVersion>'
    query += '<radioVersion>10.0.0.0</radioVersion></software>'
    query += '</clientProperties><updateDirectives><bundleVersionFilter>'
    query += '</bundleVersionFilter></updateDirectives>'
    query += '</availableBundlesRequest>'
    header = {"Content-Type": "text/xml;charset=UTF-8"}
    os.environ["REQUESTS_CA_BUNDLE"] = grab_pem()
    req = requests.post(server, headers=header, data=query)
    root = xml.etree.ElementTree.fromstring(req.text)
    package = root.find('./data/content')
    bundlelist = [child.attrib["version"] for child in package]
    return bundlelist


def ptcrb_scraper(ptcrbid):
    """
    Get the PTCRB results for a given device.

    :param ptcrbid: Numerical ID from PTCRB (end of URL).
    :type ptcrbid: str
    """
    baseurl = "https://ptcrb.com/vendor/complete/view_complete_request_guest.cfm?modelid={0}".format(
        ptcrbid)
    os.environ["REQUESTS_CA_BUNDLE"] = grab_pem()
    req = requests.get(baseurl)
    soup = BeautifulSoup(req.content, 'html.parser')
    text = soup.get_text()
    text = text.replace("\r\n", " ")
    prelimlist = re.findall("OS .+[^\\n]", text, re.IGNORECASE)
    if not prelimlist:  # Priv
        prelimlist = re.findall(r"[A-Z]{3}[0-9]{3}[\s]", text)
    cleanlist = []
    for item in prelimlist:
        if not item.endswith("\r\n"):  # they should hire QC people...
            cleanlist.append(ptcrb_item_cleaner(item))
    return cleanlist


def ptcrb_item_cleaner(item):
    """
    Cleanup poorly formatted PTCRB entries written by an intern.

    :param item: The item to clean.
    :type item: str
    """
    item = item.replace("<td>", "")
    item = item.replace("</td>", "")
    item = item.replace("\n", "")
    item = item.replace(" (SR", ", SR")
    item = re.sub(r"\s?\((.*)$", "", item)
    item = re.sub(r"\sSV.*$", "", item)
    item = item.replace(")", "")
    item = item.replace(". ", ".")
    item = item.replace(";", "")
    item = item.replace("version", "Version")
    item = item.replace("Verison", "Version")
    if item.count("OS") > 1:
        templist = item.split("OS")
        templist[0] = "OS"
        item = "".join([templist[0], templist[1]])
    item = item.replace("SR", "SW Release")
    item = item.replace(" Version:", ":")
    item = item.replace("Version ", " ")
    item = item.replace(":1", ": 1")
    item = item.replace(", ", " ")
    item = item.replace("Software", "SW")
    item = item.replace("  ", " ")
    item = item.replace("OS ", "OS: ")
    item = item.replace("Radio ", "Radio: ")
    item = item.replace("Release ", "Release: ")
    spaclist = item.split(" ")
    if len(spaclist) > 1:
        while len(spaclist[1]) < 11:
            spaclist[1] += " "
        while len(spaclist[3]) < 11:
            spaclist[3] += " "
    else:
        spaclist.insert(0, "OS:")
    item = " ".join(spaclist)
    item = item.strip()
    return item


def kernel_scraper(utils=False):
    """
    Scrape BlackBerry's GitHub kernel repo for available branches.

    :param utils: Check android-utils repo instead of android-linux-kernel. Default is False.
    :type utils: bool
    """
    repo = "android-utils" if utils else "android-linux-kernel"
    url = "https://github.com/blackberry/{0}/branches/all".format(repo)
    os.environ["REQUESTS_CA_BUNDLE"] = grab_pem()
    req = requests.get(url)
    soup = BeautifulSoup(req.content, 'html.parser')
    text = soup.get_text()
    kernlist = re.findall(r"msm[0-9]{4}\/[A-Z0-9]{6}", text, re.IGNORECASE)
    return kernlist


def make_droid_skeleton(method, build, device, variant="common"):
    """
    Make an Android autoloader/hash URL.

    :param method: None for regular OS links, "sha256/512" for SHA256 or 512 hash.
    :type method: str

    :param build: Build to check, 3 letters + 3 numbers.
    :type build: str

    :param device: Device to check.
    :type device: str

    :param variant: Autoloader variant. Default is "common".
    :type variant: str
    """
    folder = {"vzw-vzw": "verizon", "na-att": "att", "na-tmo": "tmo", "common": "default"}
    devices = {"Priv": "qc8992", "DTEK50": "qc8952_64_sfi"}
    roots = {"Priv": "bbfoundation/hashfiles_priv/{0}".format(folder[variant]), "DTEK50": "bbSupport/DTEK50"}
    base = "bbry_{2}_autoloader_user-{0}-{1}".format(variant, build.upper(), devices[device])
    if method is None:
        skel = "https://bbapps.download.blackberry.com/Priv/{0}.zip".format(base)
    else:
        skel = "http://ca.blackberry.com/content/dam/{1}/{0}.{2}sum".format(base, roots[device], method.lower())
    return skel


def droid_scanner(build, device, method=None):
    """
    Check for Android autoloaders on BlackBerry's site.

    :param build: Build to check, 3 letters + 3 numbers.
    :type build: str

    :param device: Device to check.
    :type device: str

    :param method: None for regular OS links, "sha256/512" for SHA256 or 512 hash.
    :type method: str
    """
    variants = ("common", "vzw-vzw", "na-tmo", "na-att")  # device variants
    with concurrent.futures.ThreadPoolExecutor(max_workers=len(variants)) as xec:
        results = []
        for var in variants:
            skel = make_droid_skeleton(method, build, device, var)
            avail = xec.submit(availability, skel)
            if avail.result():
                results.append(skel)
        return results if results else None


def base_metadata(url):
    """
    Get BBNDK metadata, base function.
    """
    req = requests.get(url)
    data = req.content
    entries = data.split(b"\n")
    metadata = [entry.split(b",")[1].decode("utf-8") for entry in entries if entry]
    return metadata


def ndk_metadata():
    """
    Get BBNDK target metadata.
    """
    data = base_metadata("http://downloads.blackberry.com/upr/developers/update/bbndk/metadata")
    metadata = [entry for entry in data if entry.startswith(("10.0", "10.1", "10.2"))]
    return metadata


def sim_metadata():
    """
    Get BBNDK simulator metadata.
    """
    metadata = base_metadata("http://downloads.blackberry.com/upr/developers/update/bbndk/simulator/simulator_metadata")
    return metadata


def runtime_metadata():
    """
    Get BBNDK runtime metadata.
    """
    metadata = base_metadata("http://downloads.blackberry.com/upr/developers/update/bbndk/runtime/runtime_metadata")
    return metadata


def series_generator(osversion):
    """
    Generate series/branch name from OS version.

    :param osversion: OS version.
    :type osversion: str
    """
    splits = osversion.split(".")
    return "BB{0}_{1}_{2}".format(*splits[0:3])


def devalpha_urls(osversion, skel):
    """
    Check individual Dev Alpha autoloader URLs.

    :param osversion: OS version.
    :type osversion: str

    :param skel: Individual skeleton format to try.
    :type skel: str
    """
    url = "http://downloads.blackberry.com/upr/developers/downloads/{0}{1}.exe".format(skel, osversion)
    req = requests.head(url)
    if req.status_code == 200:
        finals = (url, req.headers["content-length"])
    else:
        finals = ()
    return finals


def devalpha_urls_bootstrap(osversion, skeletons):
    """
    Get list of valid Dev Alpha autoloader URLs.

    :param osversion: OS version.
    :type osversion: str

    :param skeletons: List of skeleton formats to try.
    :type skeletons: list
    """
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as xec:
        try:
            finals = {}
            skels = skeletons
            for idx, skel in enumerate(skeletons):
                if "<SERIES>" in skel:
                    skels[idx] = skel.replace("<SERIES>", series_generator(osversion))
            for skel in skels:
                final = xec.submit(devalpha_urls, osversion, skel).result()
                if final:
                    finals[final[0]] = final[1]
            return finals
        except KeyboardInterrupt:  # pragma: no cover
            xec.shutdown(wait=False)


def dev_dupe_cleaner(finals):
    """
    Clean duplicate autoloader entries.

    :param finals: Dict of URL:content-length pairs.
    :type finals: dict(str: str)
    """
    revo = {}
    for key, val in finals.items():
        revo.setdefault(val, set()).add(key)
    dupelist = [val for key, val in revo.items() if len(val) > 1]
    for dupe in dupelist:
        for entry in dupe:
            if "DevAlpha" in entry:
                del finals[entry]
    return finals


1			#!/usr/bin/env python3
2			"""This module is used for network connections; APIs, downloading, etc."""
3
4			import os # filesystem read
5			import xml.etree.ElementTree # XML parsing
6			import re # regexes
7			import hashlib # base url creation
8			import concurrent.futures # multiprocessing/threading
9			import glob # pem file lookup
10			import requests # downloading
11			from bs4 import BeautifulSoup # scraping
12			from bbarchivist import utilities # parse filesize
13			from bbarchivist.bbconstants import SERVERS # lookup servers
14
15			__author__ = "Thurask"
16			__license__ = "WTFPL v2"
17			__copyright__ = "Copyright 2015-2016 Thurask"
18
19
20			def grab_pem():
21			"""
22			Work with either local cacerts or system cacerts. Since cx_freeze is dumb.
23			"""
24			try:
25			pemfile = glob.glob(os.path.join(os.getcwd(), "cacert.pem"))[0]
26			except IndexError:
27			return requests.certs.where() # no local cacerts
28			else:
29			return os.path.abspath(pemfile) # local cacerts
30
31
32			def get_length(url):
33			"""
34			Get content-length header from some URL.
35
36			:param url: The URL to check.
37			:type url: str
38			"""
39			os.environ["REQUESTS_CA_BUNDLE"] = grab_pem()
40			if url is None:
41			return 0
42			try:
43			heads = requests.head(url)
44			fsize = heads.headers['content-length']
45			return int(fsize)
46			except requests.ConnectionError:
47			return 0
48
49
50			def download(url, output_directory=None):
51			"""
52			Download file from given URL.
53
54			:param url: URL to download from.
55			:type url: str
56
57			:param output_directory: Download folder. Default is local.
58			:type output_directory: str
59			"""
60			if output_directory is None:
61			output_directory = os.getcwd()
62			lfname = url.split('/')[-1]
63			sname = utilities.stripper(lfname)
64			os.environ["REQUESTS_CA_BUNDLE"] = grab_pem()
65			fname = os.path.join(output_directory, lfname)
66			with open(fname, "wb") as file:
67			req = requests.get(url, stream=True)
68			clength = req.headers['content-length']
69			fsize = utilities.fsizer(clength)
70			if req.status_code == 200: # 200 OK
71			print("DOWNLOADING {0} [{1}]".format(sname, fsize))
72			for chunk in req.iter_content(chunk_size=1024):
73			file.write(chunk)
74			else:
75			print("ERROR: HTTP {0} IN {1}".format(req.status_code, lfname))
76
77
78			def download_bootstrap(urls, outdir=None, workers=5):
79			"""
80			Run downloaders for each file in given URL iterable.
81
82			:param urls: URLs to download.
83			:type urls: list
84
85			:param outdir: Download folder. Default is handled in :func:`download`.
86			:type outdir: str
87
88			:param workers: Number of worker processes. Default is 5.
89			:type workers: int
90			"""
91			workers = len(urls) if len(urls) < workers else workers
92			spinman = utilities.SpinManager()
93			with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as xec:
94			try:
95			spinman.start()
96			for url in urls:
97			xec.submit(download, url, outdir)
98			except (KeyboardInterrupt, SystemExit): # pragma: no cover
99			xec.shutdown()
100			spinman.stop()
101			spinman.stop()
102			utilities.spinner_clear()
103			utilities.line_begin()
104
105
106			def create_base_url(softwareversion):
107			"""
108			Make the root URL for production server files.
109
110			:param softwareversion: Software version to hash.
111			:type softwareversion: str
112			"""
113			# Hash software version
114			swhash = hashlib.sha1(softwareversion.encode('utf-8'))
115			hashedsoftwareversion = swhash.hexdigest()
116			# Root of all urls
117			baseurl = "http://cdn.fs.sl.blackberry.com/fs/qnx/production/" + hashedsoftwareversion
118			return baseurl
119
120
121			def availability(url):
122			"""
123			Check HTTP status code of given URL.
124			200 or 301-308 is OK, else is not.
125
126			:param url: URL to check.
127			:type url: str
128			"""
129			os.environ["REQUESTS_CA_BUNDLE"] = grab_pem()
130			try:
131			avlty = requests.head(url)
132			status = int(avlty.status_code)
133			return status == 200 or 300 < status <= 308
134			except requests.ConnectionError:
135			return False
136
137
138			def clean_availability(results, server):
139			"""
140			Clean availability for autolookup script.
141
142			:param results: Result dict.
143			:type results: dict(str: str)
144
145			:param server: Server, key for result dict.
146			:type server: str
147			"""
148			marker = "PD" if server == "p" else server.upper()
149			rel = results[server.lower()]
150			avail = marker if rel != "SR not in system" and rel is not None else " "
151			return rel, avail
152
153
154			def carrier_checker(mcc, mnc):
155			"""
156			Query BlackBerry World to map a MCC and a MNC to a country and carrier.
157
158			:param mcc: Country code.
159			:type mcc: int
160
161			:param mnc: Network code.
162			:type mnc: int
163			"""
164			url = "http://appworld.blackberry.com/ClientAPI/checkcarrier?homemcc={0}&homemnc={1}&devicevendorid=-1&pin=0".format(
165			mcc, mnc)
166			user_agent = {'User-agent': 'AppWorld/5.1.0.60'}
167			os.environ["REQUESTS_CA_BUNDLE"] = grab_pem()
168			req = requests.get(url, headers=user_agent)
169			root = xml.etree.ElementTree.fromstring(req.text)
170			for child in root:
171			if child.tag == "country":
172			country = child.get("name")
173			if child.tag == "carrier":
174			carrier = child.get("name")
175			return country, carrier
176
177
178			def return_npc(mcc, mnc):
179			"""
180			Format MCC and MNC into a NPC.
181
182			:param mcc: Country code.
183			:type mcc: int
184
185			:param mnc: Network code.
186			:type mnc: int
187			"""
188			return "{0}{1}30".format(str(mcc).zfill(3), str(mnc).zfill(3))
189
190
191			def carrier_query(npc, device, upgrade=False, blitz=False, forced=None):
192			"""
193			Query BlackBerry servers, check which update is out for a carrier.
194
195			:param npc: MCC + MNC (see `func:return_npc`)
196			:type npc: int
197
198			:param device: Hexadecimal hardware ID.
199			:type device: str
200
201			:param upgrade: Whether to use upgrade files. False by default.
202			:type upgrade: bool
203
204			:param blitz: Whether or not to create a blitz package. False by default.
205			:type blitz: bool
206
207			:param forced: Force a software release.
208			:type forced: str
209			"""
210			upg = "upgrade" if upgrade else "repair"
211			forced = "latest" if forced is None else forced
212			url = "https://cs.sl.blackberry.com/cse/updateDetails/2.2/"
213			query = '<?xml version="1.0" encoding="UTF-8"?>'
214			query += '<updateDetailRequest version="2.2.1" authEchoTS="1366644680359">'
215			query += "<clientProperties>"
216			query += "<hardware>"
217			query += "<pin>0x2FFFFFB3</pin><bsn>1128121361</bsn>"
218			query += "<imei>004401139269240</imei>"
219			query += "<id>0x{0}</id>".format(device)
220			query += "</hardware>"
221			query += "<network>"
222			query += "<homeNPC>0x{0}</homeNPC>".format(npc)
223			query += "<iccid>89014104255505565333</iccid>"
224			query += "</network>"
225			query += "<software>"
226			query += "<currentLocale>en_US</currentLocale>"
227			query += "<legalLocale>en_US</legalLocale>"
228			query += "</software>"
229			query += "</clientProperties>"
230			query += "<updateDirectives>"
231			query += '<allowPatching type="REDBEND">true</allowPatching>'
232			query += "<upgradeMode>{0}</upgradeMode>".format(upg)
233			query += "<provideDescriptions>false</provideDescriptions>"
234			query += "<provideFiles>true</provideFiles>"
235			query += "<queryType>NOTIFICATION_CHECK</queryType>"
236			query += "</updateDirectives>"
237			query += "<pollType>manual</pollType>"
238			query += "<resultPackageSetCriteria>"
239			query += '<softwareRelease softwareReleaseVersion="{0}" />'.format(forced)
240			query += "<releaseIndependent>"
241			query += '<packageType operation="include">application</packageType>'
242			query += "</releaseIndependent>"
243			query += "</resultPackageSetCriteria>"
244			query += "</updateDetailRequest>"
245			header = {"Content-Type": "text/xml;charset=UTF-8"}
246			os.environ["REQUESTS_CA_BUNDLE"] = grab_pem()
247			req = requests.post(url, headers=header, data=query)
248			return parse_carrier_xml(req.text, blitz)
249
250
251			def parse_carrier_xml(data, blitz=False):
252			"""
253			Parse the response to a carrier update request and return the juicy bits.
254
255			:param data: The data to parse.
256			:type data: xml
257
258			:param blitz: Whether or not to create a blitz package. False by default.
259			:type blitz: bool
260			"""
261			root = xml.etree.ElementTree.fromstring(data)
262			sw_exists = root.find('./data/content/softwareReleaseMetadata')
263			swver = "N/A" if sw_exists is None else ""
264			if sw_exists is not None:
265			for child in root.iter("softwareReleaseMetadata"):
266			swver = child.get("softwareReleaseVersion")
267			files = []
268			osver = ""
269			radver = ""
270			package_exists = root.find('./data/content/fileSets/fileSet')
271			if package_exists is not None:
272			baseurl = "{0}/".format(package_exists.get("url"))
273			for child in root.iter("package"):
274			if not blitz:
275			files.append(baseurl + child.get("path"))
276			else:
277			if child.get("type") not in ["system:radio", "system:desktop", "system:os"]:
278			files.append(baseurl + child.get("path"))
279			if child.get("type") == "system:radio":
280			radver = child.get("version")
281			if child.get("type") == "system:desktop":
282			osver = child.get("version")
283			if child.get("type") == "system:os":
284			osver = child.get("version")
285			else:
286			pass
287			return(swver, osver, radver, files)
288
289
290			def sr_lookup(osver, server):
291			"""
292			Software release lookup, with choice of server.
293			:data:`bbarchivist.bbconstants.SERVERLIST` for server list.
294
295			:param osver: OS version to lookup, 10.x.y.zzzz.
296			:type osver: str
297
298			:param server: Server to use.
299			:type server: str
300			"""
301			reg = re.compile(r"(\d{1,4}\.)(\d{1,4}\.)(\d{1,4}\.)(\d{1,4})")
302			query = '<?xml version="1.0" encoding="UTF-8"?>'
303			query += '<srVersionLookupRequest version="2.0.0"'
304			query += ' authEchoTS="1366644680359">'
305			query += '<clientProperties><hardware>'
306			query += '<pin>0x2FFFFFB3</pin><bsn>1140011878</bsn>'
307			query += '<imei>004402242176786</imei><id>0x8D00240A</id>'
308			query += '<isBootROMSecure>true</isBootROMSecure>'
309			query += '</hardware>'
310			query += '<network>'
311			query += '<vendorId>0x0</vendorId><homeNPC>0x60</homeNPC>'
312			query += '<currentNPC>0x60</currentNPC><ecid>0x1</ecid>'
313			query += '</network>'
314			query += '<software><currentLocale>en_US</currentLocale>'
315			query += '<legalLocale>en_US</legalLocale>'
316			query += '<osVersion>{0}</osVersion>'.format(osver)
317			query += '<omadmEnabled>false</omadmEnabled>'
318			query += '</software></clientProperties>'
319			query += '</srVersionLookupRequest>'
320			header = {"Content-Type": "text/xml;charset=UTF-8"}
321			os.environ["REQUESTS_CA_BUNDLE"] = grab_pem()
322			try:
323			req = requests.post(server, headers=header, data=query, timeout=1)
324			except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
325			return "SR not in system"
326			try:
327			root = xml.etree.ElementTree.fromstring(req.text)
328			except xml.etree.ElementTree.ParseError:
329			return "SR not in system"
330			else:
331			packages = root.findall('./data/content/')
332			for package in packages:
333			if package.text is not None:
334			match = reg.match(package.text)
335			if match:
336			return package.text
337			else:
338			return "SR not in system"
339
340
341			def sr_lookup_bootstrap(osv):
342			"""
343			Run lookups for each server for given OS.
344
345			:param osv: OS to check.
346			:type osv: str
347			"""
348			with concurrent.futures.ThreadPoolExecutor(max_workers=5) as xec:
349			try:
350			results = {
351			"p": None,
352			"a1": None,
353			"a2": None,
354			"b1": None,
355			"b2": None
356			}
357			for key in results:
358			results[key] = xec.submit(sr_lookup, osv, SERVERS[key]).result()
359			return results
360			except KeyboardInterrupt: # pragma: no cover
361			xec.shutdown(wait=False)
362
363
364			def available_bundle_lookup(mcc, mnc, device):
365			"""
366			Check which software releases were ever released for a carrier.
367
368			:param mcc: Country code.
369			:type mcc: int
370
371			:param mnc: Network code.
372			:type mnc: int
373
374			:param device: Hexadecimal hardware ID.
375			:type device: str
376			"""
377			server = "https://cs.sl.blackberry.com/cse/availableBundles/1.0.0/"
378			npc = return_npc(mcc, mnc)
379			query = '<?xml version="1.0" encoding="UTF-8"?>'
380			query += '<availableBundlesRequest version="1.0.0" '
381			query += 'authEchoTS="1366644680359">'
382			query += '<deviceId><pin>0x2FFFFFB3</pin></deviceId>'
383			query += '<clientProperties><hardware><id>0x{0}</id>'.format(device)
384			query += '<isBootROMSecure>true</isBootROMSecure></hardware>'
385			query += '<network><vendorId>0x0</vendorId><homeNPC>0x{0}</homeNPC>'.format(npc)
386			query += '<currentNPC>0x{0}</currentNPC></network><software>'.format(npc)
387			query += '<currentLocale>en_US</currentLocale>'
388			query += '<legalLocale>en_US</legalLocale>'
389			query += '<osVersion>10.0.0.0</osVersion>'
390			query += '<radioVersion>10.0.0.0</radioVersion></software>'
391			query += '</clientProperties><updateDirectives><bundleVersionFilter>'
392			query += '</bundleVersionFilter></updateDirectives>'
393			query += '</availableBundlesRequest>'
394			header = {"Content-Type": "text/xml;charset=UTF-8"}
395			os.environ["REQUESTS_CA_BUNDLE"] = grab_pem()
396			req = requests.post(server, headers=header, data=query)
397			root = xml.etree.ElementTree.fromstring(req.text)
398			package = root.find('./data/content')
399			bundlelist = [child.attrib["version"] for child in package]
400			return bundlelist
401
402
403			def ptcrb_scraper(ptcrbid):
404			"""
405			Get the PTCRB results for a given device.
406
407			:param ptcrbid: Numerical ID from PTCRB (end of URL).
408			:type ptcrbid: str
409			"""
410			baseurl = "https://ptcrb.com/vendor/complete/view_complete_request_guest.cfm?modelid={0}".format(
411			ptcrbid)
412			os.environ["REQUESTS_CA_BUNDLE"] = grab_pem()
413			req = requests.get(baseurl)
414			soup = BeautifulSoup(req.content, 'html.parser')
415			text = soup.get_text()
416			text = text.replace("\r\n", " ")
417			prelimlist = re.findall("OS .+[^\\n]", text, re.IGNORECASE)
418			if not prelimlist: # Priv
419			prelimlist = re.findall(r"[A-Z]{3}[0-9]{3}[\s]", text)
420			cleanlist = []
421			for item in prelimlist:
422			if not item.endswith("\r\n"): # they should hire QC people...
423			cleanlist.append(ptcrb_item_cleaner(item))
424			return cleanlist
425
426
427			def ptcrb_item_cleaner(item):
428			"""
429			Cleanup poorly formatted PTCRB entries written by an intern.
430
431			:param item: The item to clean.
432			:type item: str
433			"""
434			item = item.replace("<td>", "")
435			item = item.replace("</td>", "")
436			item = item.replace("\n", "")
437			item = item.replace(" (SR", ", SR")
438			item = re.sub(r"\s?\((.*)$", "", item)
439			item = re.sub(r"\sSV.*$", "", item)
440			item = item.replace(")", "")
441			item = item.replace(". ", ".")
442			item = item.replace(";", "")
443			item = item.replace("version", "Version")
444			item = item.replace("Verison", "Version")
445			if item.count("OS") > 1:
446			templist = item.split("OS")
447			templist[0] = "OS"
448			item = "".join([templist[0], templist[1]])
449			item = item.replace("SR", "SW Release")
450			item = item.replace(" Version:", ":")
451			item = item.replace("Version ", " ")
452			item = item.replace(":1", ": 1")
453			item = item.replace(", ", " ")
454			item = item.replace("Software", "SW")
455			item = item.replace(" ", " ")
456			item = item.replace("OS ", "OS: ")
457			item = item.replace("Radio ", "Radio: ")
458			item = item.replace("Release ", "Release: ")
459			spaclist = item.split(" ")
460			if len(spaclist) > 1:
461			while len(spaclist[1]) < 11:
462			spaclist[1] += " "
463			while len(spaclist[3]) < 11:
464			spaclist[3] += " "
465			else:
466			spaclist.insert(0, "OS:")
467			item = " ".join(spaclist)
468			item = item.strip()
469			return item
470
471
472			def kernel_scraper(utils=False):
473			"""
474			Scrape BlackBerry's GitHub kernel repo for available branches.
475
476			:param utils: Check android-utils repo instead of android-linux-kernel. Default is False.
477			:type utils: bool
478			"""
479			repo = "android-utils" if utils else "android-linux-kernel"
480			url = "https://github.com/blackberry/{0}/branches/all".format(repo)
481			os.environ["REQUESTS_CA_BUNDLE"] = grab_pem()
482			req = requests.get(url)
483			soup = BeautifulSoup(req.content, 'html.parser')
484			text = soup.get_text()
485			kernlist = re.findall(r"msm[0-9]{4}\/[A-Z0-9]{6}", text, re.IGNORECASE)
486			return kernlist
487
488
489			def make_droid_skeleton(method, build, device, variant="common"):
490			"""
491			Make an Android autoloader/hash URL.
492
493			:param method: None for regular OS links, "sha256/512" for SHA256 or 512 hash.
494			:type method: str
495
496			:param build: Build to check, 3 letters + 3 numbers.
497			:type build: str
498
499			:param device: Device to check.
500			:type device: str
501
502			:param variant: Autoloader variant. Default is "common".
503			:type variant: str
504			"""
505			folder = {"vzw-vzw": "verizon", "na-att": "att", "na-tmo": "tmo", "common": "default"}
506			devices = {"Priv": "qc8992", "DTEK50": "qc8952_64_sfi"}
507			roots = {"Priv": "bbfoundation/hashfiles_priv/{0}".format(folder[variant]), "DTEK50": "bbSupport/DTEK50"}
508			base = "bbry_{2}_autoloader_user-{0}-{1}".format(variant, build.upper(), devices[device])
509			if method is None:
510			skel = "https://bbapps.download.blackberry.com/Priv/{0}.zip".format(base)
511			else:
512			skel = "http://ca.blackberry.com/content/dam/{1}/{0}.{2}sum".format(base, roots[device], method.lower())
513			return skel
514
515
516			def droid_scanner(build, device, method=None):
517			"""
518			Check for Android autoloaders on BlackBerry's site.
519
520			:param build: Build to check, 3 letters + 3 numbers.
521			:type build: str
522
523			:param device: Device to check.
524			:type device: str
525
526			:param method: None for regular OS links, "sha256/512" for SHA256 or 512 hash.
527			:type method: str
528			"""
529			variants = ("common", "vzw-vzw", "na-tmo", "na-att") # device variants
530			with concurrent.futures.ThreadPoolExecutor(max_workers=len(variants)) as xec:
531			results = []
532			for var in variants:
533			skel = make_droid_skeleton(method, build, device, var)
534			avail = xec.submit(availability, skel)
535			if avail.result():
536			results.append(skel)
537			return results if results else None
538
539
540			def base_metadata(url):
541			"""
542			Get BBNDK metadata, base function.
543			"""
544			req = requests.get(url)
545			data = req.content
546			entries = data.split(b"\n")
547			metadata = [entry.split(b",")[1].decode("utf-8") for entry in entries if entry]
548			return metadata
549
550
551			def ndk_metadata():
552			"""
553			Get BBNDK target metadata.
554			"""
555			data = base_metadata("http://downloads.blackberry.com/upr/developers/update/bbndk/metadata")
556			metadata = [entry for entry in data if entry.startswith(("10.0", "10.1", "10.2"))]
557			return metadata
558
559
560			def sim_metadata():
561			"""
562			Get BBNDK simulator metadata.
563			"""
564			metadata = base_metadata("http://downloads.blackberry.com/upr/developers/update/bbndk/simulator/simulator_metadata")
565			return metadata
566
567
568			def runtime_metadata():
569			"""
570			Get BBNDK runtime metadata.
571			"""
572			metadata = base_metadata("http://downloads.blackberry.com/upr/developers/update/bbndk/runtime/runtime_metadata")
573			return metadata
574
575
576			def series_generator(osversion):
577			"""
578			Generate series/branch name from OS version.
579
580			:param osversion: OS version.
581			:type osversion: str
582			"""
583			splits = osversion.split(".")
584			return "BB{0}_{1}_{2}".format(*splits[0:3])
585
586
587			def devalpha_urls(osversion, skel):
588			"""
589			Check individual Dev Alpha autoloader URLs.
590
591			:param osversion: OS version.
592			:type osversion: str
593
594			:param skel: Individual skeleton format to try.
595			:type skel: str
596			"""
597			url = "http://downloads.blackberry.com/upr/developers/downloads/{0}{1}.exe".format(skel, osversion)
598			req = requests.head(url)
599			if req.status_code == 200:
600			finals = (url, req.headers["content-length"])
601			else:
602			finals = ()
603			return finals
604
605
606			def devalpha_urls_bootstrap(osversion, skeletons):
607			"""
608			Get list of valid Dev Alpha autoloader URLs.
609
610			:param osversion: OS version.
611			:type osversion: str
612
613			:param skeletons: List of skeleton formats to try.
614			:type skeletons: list
615			"""
616			with concurrent.futures.ThreadPoolExecutor(max_workers=5) as xec:
617			try:
618			finals = {}
619			skels = skeletons
620			for idx, skel in enumerate(skeletons):
621			if "<SERIES>" in skel:
622			skels[idx] = skel.replace("<SERIES>", series_generator(osversion))
623			for skel in skels:
624			final = xec.submit(devalpha_urls, osversion, skel).result()
625			if final:
626			finals[final[0]] = final[1]
627			return finals
628			except KeyboardInterrupt: # pragma: no cover
629			xec.shutdown(wait=False)
630
631
632			def dev_dupe_cleaner(finals):
633			"""
634			Clean duplicate autoloader entries.
635
636			:param finals: Dict of URL:content-length pairs.
637			:type finals: dict(str: str)
638			"""
639			revo = {}
640			for key, val in finals.items():
641			revo.setdefault(val, set()).add(key)
642			dupelist = [val for key, val in revo.items() if len(val) > 1]
643			for dupe in dupelist:
644			for entry in dupe:
645			if "DevAlpha" in entry:
646			del finals[entry]
647			return finals
648

thurask / bbarchivist

Push — master ( 473d55...344b16 )

make_priv_skeleton() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like