bbarchivist.xmlutils.parse_sr_lookup() - Code Metrics - Inspection of "move xml handlers to new file" - thurask/bbarchivist - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 57fbcd...e67727 )

by John

created 2018-02-07 03:19 UTC

bbarchivist.xmlutils.parse_sr_lookup() A

↳ Parent: bbarchivist.xmlutils

Complexity

Conditions

Size

Total Lines	14
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	7
CRAP Score	3

Importance

Changes

Metric	Value
cc	3
eloc	7
nop	1
dl	0
loc	14
ccs	7
cts	7
cp	1
crap	3
rs	9.4285
c	0
b	0
f	0

#!/usr/bin/env python3
"""This module is used for XML handling."""

import os  # filesystem read
import random  # choice
import re  # regexes

try:
    from defusedxml import ElementTree  # safer XML parsing
except (ImportError, AttributeError):
    from xml.etree import ElementTree  # XML parsing

__author__ = "Thurask"
__license__ = "WTFPL v2"
__copyright__ = "2018 Thurask"


def parse_tcl_check(data):
    """
    Extract version and file info from TCL update server response.

    :param data: The data to parse.
    :type data: str
    """
    root = ElementTree.fromstring(data)
    tvver = root.find("VERSION").find("TV").text
    fwid = root.find("FIRMWARE").find("FW_ID").text
    fileinfo = root.find("FIRMWARE").find("FILESET").find("FILE")
    filename = fileinfo.find("FILENAME").text
    filesize = fileinfo.find("SIZE").text
    filehash = fileinfo.find("CHECKSUM").text
    return tvver, fwid, filename, filesize, filehash


def dump_tcl_xml(xmldata, salt):
    """
    Write XML responses to output directory.

    :param xmldata: Response XML.
    :type xmldata: str

    :param salt: Salt hash.
    :type salt: str
    """
    outfile = os.path.join(os.getcwd(), "logs", "{0}.xml".format(salt))
    if not os.path.exists(os.path.dirname(outfile)):
        os.makedirs(os.path.dirname(outfile))
    with open(outfile, "w", encoding="utf-8") as afile:
        afile.write(xmldata)


def parse_tcl_download_request(body, mode=4):
    """
    Extract file URL and encrypt slave URL from TCL update server response.

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

    :param mode: 4 if downloading autoloaders, 2 if downloading OTA deltas.
    :type mode: int
    """
    root = ElementTree.fromstring(body)
    slavelist = root.find("SLAVE_LIST").findall("SLAVE")
    slave = random.choice(slavelist).text
    dlurl = root.find("FILE_LIST").find("FILE").find("DOWNLOAD_URL").text
    eslave = root.find("SLAVE_LIST").findall("ENCRYPT_SLAVE")
    encslave = None if mode == 2 or not eslave else random.choice(eslave).text
    return "http://{0}{1}".format(slave, dlurl), encslave


def cchecker_get_tags(roottext):
    """
    Get country and carrier from XML.

    :param roottext: XML text.
    :type roottext: str
    """
    root = ElementTree.fromstring(roottext)
    for child in root:
        if child.tag == "country":
            country = child.get("name")
        if child.tag == "carrier":
            carrier = child.get("name")
    return country, carrier


def prep_available_bundle(device, npc):
    """
    Prepare bundle query XML.

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

    :param npc: MCC + MNC (see `func:bbarchivist.networkutils.return_npc`)
    :type npc: int
    """
    query = '<?xml version="1.0" encoding="UTF-8"?><availableBundlesRequest version="1.0.0" authEchoTS="1366644680359"><deviceId><pin>0x2FFFFFB3</pin></deviceId><clientProperties><hardware><id>0x{0}</id><isBootROMSecure>true</isBootROMSecure></hardware><network><vendorId>0x0</vendorId><homeNPC>0x{1}</homeNPC><currentNPC>0x{1}</currentNPC></network><software><currentLocale>en_US</currentLocale><legalLocale>en_US</legalLocale><osVersion>10.0.0.0</osVersion><radioVersion>10.0.0.0</radioVersion></software></clientProperties><updateDirectives><bundleVersionFilter></bundleVersionFilter></updateDirectives></availableBundlesRequest>'.format(device, npc)
    return query


def parse_available_bundle(roottext):
    """
    Get bundles from XML.

    :param roottext: XML text.
    :type roottext: str
    """
    root = ElementTree.fromstring(roottext)
    package = root.find('./data/content')
    bundlelist = [child.attrib["version"] for child in package]
    return bundlelist


def carrier_swver_get(root):
    """
    Get software release from carrier XML.

    :param root: ElementTree we're barking up.
    :type root: xml.etree.ElementTree.ElementTree
    """
    for child in root.iter("softwareReleaseMetadata"):
        swver = child.get("softwareReleaseVersion")
    return swver


def carrier_child_fileappend(child, files, baseurl, blitz=False):
    """
    Append bar file links to a list from a child element.

    :param child: Child element in use.
    :type child: xml.etree.ElementTree.Element

    :param files: Filelist.
    :type files: list(str)

    :param baseurl: Base URL, URL minus the filename.
    :type baseurl: str

    :param blitz: Whether or not to create a blitz package. False by default.
    :type blitz: bool
    """
    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"))
    return files


def carrier_child_finder(root, files, baseurl, blitz=False):
    """
    Extract filenames, radio and OS from child elements.

    :param root: ElementTree we're barking up.
    :type root: xml.etree.ElementTree.ElementTree

    :param files: Filelist.
    :type files: list(str)

    :param baseurl: Base URL, URL minus the filename.
    :type baseurl: str

    :param blitz: Whether or not to create a blitz package. False by default.
    :type blitz: bool
    """
    osver = radver = ""
    for child in root.iter("package"):
        files = carrier_child_fileappend(child, files, baseurl, blitz)
        if child.get("type") == "system:radio":
            radver = child.get("version")
        elif child.get("type") == "system:desktop":
            osver = child.get("version")
        elif child.get("type") == "system:os":
            osver = child.get("version")
    return osver, radver, files


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 = 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:
        swver = carrier_swver_get(root)
    files = []
    package_exists = root.find('./data/content/fileSets/fileSet')
    osver = radver = ""
    if package_exists is not None:
        baseurl = "{0}/".format(package_exists.get("url"))
        osver, radver, files = carrier_child_finder(root, files, baseurl, blitz)
    return swver, osver, radver, files


def prep_carrier_query(npc, device, upg, forced):
    """
    Prepare carrier query XML.

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

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

    :param upg: "upgrade" or "repair".
    :type upg: str

    :param forced: Force a software release.
    :type forced: str
    """
    query = '<?xml version="1.0" encoding="UTF-8"?><updateDetailRequest version="2.2.1" authEchoTS="1366644680359"><clientProperties><hardware><pin>0x2FFFFFB3</pin><bsn>1128121361</bsn><imei>004401139269240</imei><id>0x{0}</id></hardware><network><homeNPC>0x{1}</homeNPC><iccid>89014104255505565333</iccid></network><software><currentLocale>en_US</currentLocale><legalLocale>en_US</legalLocale></software></clientProperties><updateDirectives><allowPatching type="REDBEND">true</allowPatching><upgradeMode>{2}</upgradeMode><provideDescriptions>false</provideDescriptions><provideFiles>true</provideFiles><queryType>NOTIFICATION_CHECK</queryType></updateDirectives><pollType>manual</pollType><resultPackageSetCriteria><softwareRelease softwareReleaseVersion="{3}" /><releaseIndependent><packageType operation="include">application</packageType></releaseIndependent></resultPackageSetCriteria></updateDetailRequest>'.format(device, npc, upg, forced)
    return query


def prep_sr_lookup(osver):
    """
    Prepare software lookup XML.

    :param osver: OS version to lookup, 10.x.y.zzzz.
    :type osver: str
    """
    query = '<?xml version="1.0" encoding="UTF-8"?><srVersionLookupRequest version="2.0.0" authEchoTS="1366644680359"><clientProperties><hardware><pin>0x2FFFFFB3</pin><bsn>1140011878</bsn><imei>004402242176786</imei><id>0x8D00240A</id><isBootROMSecure>true</isBootROMSecure></hardware><network><vendorId>0x0</vendorId><homeNPC>0x60</homeNPC><currentNPC>0x60</currentNPC><ecid>0x1</ecid></network><software><currentLocale>en_US</currentLocale><legalLocale>en_US</legalLocale><osVersion>{0}</osVersion><omadmEnabled>false</omadmEnabled></software></clientProperties></srVersionLookupRequest>'.format(osver)
    return query

def parse_sr_lookup(reqtext):
    """
    Take the text of a software lookup request response and parse it as XML.

    :param reqtext: Response text, hopefully XML formatted.
    :type reqtext: str
    """
    try:
        root = ElementTree.fromstring(reqtext)
    except ElementTree.ParseError:
        packtext = "SR not in system"
    else:
        packtext = sr_lookup_extractor(root)
    return packtext


def sr_lookup_extractor(root):
    """
    Take an ElementTree and extract a software release from it.

    :param root: ElementTree we're barking up.
    :type root: xml.etree.ElementTree.ElementTree
    """
    reg = re.compile(r"(\d{1,4}\.)(\d{1,4}\.)(\d{1,4}\.)(\d{1,4})")
    packages = root.findall('./data/content/')
    for package in packages:
        if package.text is not None:
            match = reg.match(package.text)
            packtext = package.text if match else "SR not in system"
            return packtext


1		#!/usr/bin/env python3
2	5	"""This module is used for XML handling."""
3
4	5	import os # filesystem read
5	5	import random # choice
6	5	import re # regexes
7
8	5	try:
9	5	from defusedxml import ElementTree # safer XML parsing
10	1	except (ImportError, AttributeError):
11	1	from xml.etree import ElementTree # XML parsing
12
13	5	__author__ = "Thurask"
14	5	__license__ = "WTFPL v2"
15	5	__copyright__ = "2018 Thurask"
16
17
18	5	def parse_tcl_check(data):
19		"""
20		Extract version and file info from TCL update server response.
21
22		:param data: The data to parse.
23		:type data: str
24		"""
25	5	root = ElementTree.fromstring(data)
26	5	tvver = root.find("VERSION").find("TV").text
27	5	fwid = root.find("FIRMWARE").find("FW_ID").text
28	5	fileinfo = root.find("FIRMWARE").find("FILESET").find("FILE")
29	5	filename = fileinfo.find("FILENAME").text
30	5	filesize = fileinfo.find("SIZE").text
31	5	filehash = fileinfo.find("CHECKSUM").text
32	5	return tvver, fwid, filename, filesize, filehash
33
34
35	5	def dump_tcl_xml(xmldata, salt):
36		"""
37		Write XML responses to output directory.
38
39		:param xmldata: Response XML.
40		:type xmldata: str
41
42		:param salt: Salt hash.
43		:type salt: str
44		"""
45	5	outfile = os.path.join(os.getcwd(), "logs", "{0}.xml".format(salt))
46	5	if not os.path.exists(os.path.dirname(outfile)):
47	5	os.makedirs(os.path.dirname(outfile))
48	5	with open(outfile, "w", encoding="utf-8") as afile:
49	5	afile.write(xmldata)
50
51
52	5	def parse_tcl_download_request(body, mode=4):
53		"""
54		Extract file URL and encrypt slave URL from TCL update server response.
55
56		:param data: The data to parse.
57		:type data: str
58
59		:param mode: 4 if downloading autoloaders, 2 if downloading OTA deltas.
60		:type mode: int
61		"""
62	5	root = ElementTree.fromstring(body)
63	5	slavelist = root.find("SLAVE_LIST").findall("SLAVE")
64	5	slave = random.choice(slavelist).text
65	5	dlurl = root.find("FILE_LIST").find("FILE").find("DOWNLOAD_URL").text
66	5	eslave = root.find("SLAVE_LIST").findall("ENCRYPT_SLAVE")
67	5	encslave = None if mode == 2 or not eslave else random.choice(eslave).text
68	5	return "http://{0}{1}".format(slave, dlurl), encslave
69
70
71	5	def cchecker_get_tags(roottext):
72		"""
73		Get country and carrier from XML.
74
75		:param roottext: XML text.
76		:type roottext: str
77		"""
78	5	root = ElementTree.fromstring(roottext)
79	5	for child in root:
80	5	if child.tag == "country":
81	5	country = child.get("name")
82	5	if child.tag == "carrier":
83	5	carrier = child.get("name")
84	5	return country, carrier
85
86
87	5	def prep_available_bundle(device, npc):
88		"""
89		Prepare bundle query XML.
90
91		:param device: Hexadecimal hardware ID.
92		:type device: str
93
94		:param npc: MCC + MNC (see `func:bbarchivist.networkutils.return_npc`)
95		:type npc: int
96		"""
97	5	query = '<?xml version="1.0" encoding="UTF-8"?><availableBundlesRequest version="1.0.0" authEchoTS="1366644680359"><deviceId><pin>0x2FFFFFB3</pin></deviceId><clientProperties><hardware><id>0x{0}</id><isBootROMSecure>true</isBootROMSecure></hardware><network><vendorId>0x0</vendorId><homeNPC>0x{1}</homeNPC><currentNPC>0x{1}</currentNPC></network><software><currentLocale>en_US</currentLocale><legalLocale>en_US</legalLocale><osVersion>10.0.0.0</osVersion><radioVersion>10.0.0.0</radioVersion></software></clientProperties><updateDirectives><bundleVersionFilter></bundleVersionFilter></updateDirectives></availableBundlesRequest>'.format(device, npc)
98	5	return query
99
100
101	5	def parse_available_bundle(roottext):
102		"""
103		Get bundles from XML.
104
105		:param roottext: XML text.
106		:type roottext: str
107		"""
108	5	root = ElementTree.fromstring(roottext)
109	5	package = root.find('./data/content')
110	5	bundlelist = [child.attrib["version"] for child in package]
111	5	return bundlelist
112
113
114	5	def carrier_swver_get(root):
115		"""
116		Get software release from carrier XML.
117
118		:param root: ElementTree we're barking up.
119		:type root: xml.etree.ElementTree.ElementTree
120		"""
121	5	for child in root.iter("softwareReleaseMetadata"):
122	5	swver = child.get("softwareReleaseVersion")
123	5	return swver
124
125
126	5	def carrier_child_fileappend(child, files, baseurl, blitz=False):
127		"""
128		Append bar file links to a list from a child element.
129
130		:param child: Child element in use.
131		:type child: xml.etree.ElementTree.Element
132
133		:param files: Filelist.
134		:type files: list(str)
135
136		:param baseurl: Base URL, URL minus the filename.
137		:type baseurl: str
138
139		:param blitz: Whether or not to create a blitz package. False by default.
140		:type blitz: bool
141		"""
142	5	if not blitz:
143	5	files.append(baseurl + child.get("path"))
144		else:
145	5	if child.get("type") not in ["system:radio", "system:desktop", "system:os"]:
146	5	files.append(baseurl + child.get("path"))
147	5	return files
148
149
150	5	def carrier_child_finder(root, files, baseurl, blitz=False):
151		"""
152		Extract filenames, radio and OS from child elements.
153
154		:param root: ElementTree we're barking up.
155		:type root: xml.etree.ElementTree.ElementTree
156
157		:param files: Filelist.
158		:type files: list(str)
159
160		:param baseurl: Base URL, URL minus the filename.
161		:type baseurl: str
162
163		:param blitz: Whether or not to create a blitz package. False by default.
164		:type blitz: bool
165		"""
166	5	osver = radver = ""
167	5	for child in root.iter("package"):
168	5	files = carrier_child_fileappend(child, files, baseurl, blitz)
169	5	if child.get("type") == "system:radio":
170	5	radver = child.get("version")
171	5	elif child.get("type") == "system:desktop":
172	5	osver = child.get("version")
173	5	elif child.get("type") == "system:os":
174	5	osver = child.get("version")
175	5	return osver, radver, files
176
177
178	5	def parse_carrier_xml(data, blitz=False):
179		"""
180		Parse the response to a carrier update request and return the juicy bits.
181
182		:param data: The data to parse.
183		:type data: xml
184
185		:param blitz: Whether or not to create a blitz package. False by default.
186		:type blitz: bool
187		"""
188	5	root = ElementTree.fromstring(data)
189	5	sw_exists = root.find('./data/content/softwareReleaseMetadata')
190	5	swver = "N/A" if sw_exists is None else ""
191	5	if sw_exists is not None:
192	5	swver = carrier_swver_get(root)
193	5	files = []
194	5	package_exists = root.find('./data/content/fileSets/fileSet')
195	5	osver = radver = ""
196	5	if package_exists is not None:
197	5	baseurl = "{0}/".format(package_exists.get("url"))
198	5	osver, radver, files = carrier_child_finder(root, files, baseurl, blitz)
199	5	return swver, osver, radver, files
200
201
202	5	def prep_carrier_query(npc, device, upg, forced):
203		"""
204		Prepare carrier query XML.
205
206		:param npc: MCC + MNC (see `func:return_npc`)
207		:type npc: int
208
209		:param device: Hexadecimal hardware ID.
210		:type device: str
211
212		:param upg: "upgrade" or "repair".
213		:type upg: str
214
215		:param forced: Force a software release.
216		:type forced: str
217		"""
218	5	query = '<?xml version="1.0" encoding="UTF-8"?><updateDetailRequest version="2.2.1" authEchoTS="1366644680359"><clientProperties><hardware><pin>0x2FFFFFB3</pin><bsn>1128121361</bsn><imei>004401139269240</imei><id>0x{0}</id></hardware><network><homeNPC>0x{1}</homeNPC><iccid>89014104255505565333</iccid></network><software><currentLocale>en_US</currentLocale><legalLocale>en_US</legalLocale></software></clientProperties><updateDirectives><allowPatching type="REDBEND">true</allowPatching><upgradeMode>{2}</upgradeMode><provideDescriptions>false</provideDescriptions><provideFiles>true</provideFiles><queryType>NOTIFICATION_CHECK</queryType></updateDirectives><pollType>manual</pollType><resultPackageSetCriteria><softwareRelease softwareReleaseVersion="{3}" /><releaseIndependent><packageType operation="include">application</packageType></releaseIndependent></resultPackageSetCriteria></updateDetailRequest>'.format(device, npc, upg, forced)
219	5	return query
220
221
222	5	def prep_sr_lookup(osver):
223		"""
224		Prepare software lookup XML.
225
226		:param osver: OS version to lookup, 10.x.y.zzzz.
227		:type osver: str
228		"""
229	5	query = '<?xml version="1.0" encoding="UTF-8"?><srVersionLookupRequest version="2.0.0" authEchoTS="1366644680359"><clientProperties><hardware><pin>0x2FFFFFB3</pin><bsn>1140011878</bsn><imei>004402242176786</imei><id>0x8D00240A</id><isBootROMSecure>true</isBootROMSecure></hardware><network><vendorId>0x0</vendorId><homeNPC>0x60</homeNPC><currentNPC>0x60</currentNPC><ecid>0x1</ecid></network><software><currentLocale>en_US</currentLocale><legalLocale>en_US</legalLocale><osVersion>{0}</osVersion><omadmEnabled>false</omadmEnabled></software></clientProperties></srVersionLookupRequest>'.format(osver)
230	5	return query
231
232	5	def parse_sr_lookup(reqtext):
233		"""
234		Take the text of a software lookup request response and parse it as XML.
235
236		:param reqtext: Response text, hopefully XML formatted.
237		:type reqtext: str
238		"""
239	5	try:
240	5	root = ElementTree.fromstring(reqtext)
241	5	except ElementTree.ParseError:
242	5	packtext = "SR not in system"
243		else:
244	5	packtext = sr_lookup_extractor(root)
245	5	return packtext
246
247
248	5	def sr_lookup_extractor(root):
249		"""
250		Take an ElementTree and extract a software release from it.
251
252		:param root: ElementTree we're barking up.
253		:type root: xml.etree.ElementTree.ElementTree
254		"""
255	5	reg = re.compile(r"(\d{1,4}\.)(\d{1,4}\.)(\d{1,4}\.)(\d{1,4})")
256	5	packages = root.findall('./data/content/')
257	5	for package in packages:
258	5	if package.text is not None:
259	5	match = reg.match(package.text)
260	5	packtext = package.text if match else "SR not in system"
261		return packtext
262

thurask / bbarchivist

Push — master ( 57fbcd...e67727 )

bbarchivist.xmlutils.parse_sr_lookup() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like