pem_wrapper() - Code Metrics - Inspection of "make REQUESTS_CA_BUNDLE a decorator" - thurask/bbarchivist - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( b67795...a6aecf )

by John

created 2016-08-13 15:46 UTC

pem_wrapper() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

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

#!/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 pem_wrapper(method):
    """
    Decorator to set REQUESTS_CA_BUNDLE.

    :param method: Method to use.
    :type method: function
    """
    def wrapper(*args, **kwargs):
        """
        Set REQUESTS_CA_BUNDLE before doing function.
        """
        os.environ["REQUESTS_CA_BUNDLE"] = grab_pem()
        return method(*args, **kwargs)
    return wrapper


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

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


@pem_wrapper
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)
    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


@pem_wrapper
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
    """
    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


@pem_wrapper
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'}
    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))


@pem_wrapper
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"}
    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)


@pem_wrapper
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"}
    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)


@pem_wrapper
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"}
    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


@pem_wrapper
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)
    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


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

thurask / bbarchivist

Push — master ( b67795...a6aecf )

pem_wrapper() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like