From af49985dd1a8e7be6632ef1125f049823418f419 Mon Sep 17 00:00:00 2001 From: Ajin Abraham Date: Sun, 24 Nov 2024 22:53:21 -0800 Subject: [PATCH] [4.2.7] Androguard & ApkInspector Bump + Patch AXMLParsing (#2461) * Androguard 293ab2d89ab9ce011c7dbbc5df3c876172875a1c update * AXML Parser warn "reserved must be zero!" instead of raise * Fallback on get app name when androguard returns empty string --- LICENSES/androguard.txt | 200 ++- mobsf/MobSF/init.py | 2 +- mobsf/StaticAnalyzer/tools/androguard4/apk.py | 1477 ++++++++++++---- .../tools/androguard4/apkinspector/extract.py | 106 ++ .../{zipfile.py => apkinspector/headers.py} | 160 +- .../tools/androguard4/apkinspector/helpers.py | 68 + .../StaticAnalyzer/tools/androguard4/axml.py | 1553 ++++++++++++----- .../tools/androguard4/resources/public.py | 9 +- .../StaticAnalyzer/tools/androguard4/types.py | 14 +- .../StaticAnalyzer/tools/androguard4/util.py | 64 + mobsf/StaticAnalyzer/views/android/apk.py | 2 +- mobsf/StaticAnalyzer/views/android/app.py | 6 +- .../views/android/cert_analysis.py | 8 +- mobsf/StaticAnalyzer/views/android/jar_aar.py | 2 +- .../views/android/manifest_utils.py | 2 +- pyproject.toml | 2 +- 16 files changed, 2672 insertions(+), 1003 deletions(-) create mode 100644 mobsf/StaticAnalyzer/tools/androguard4/apkinspector/extract.py rename mobsf/StaticAnalyzer/tools/androguard4/{zipfile.py => apkinspector/headers.py} (85%) create mode 100644 mobsf/StaticAnalyzer/tools/androguard4/apkinspector/helpers.py create mode 100644 mobsf/StaticAnalyzer/tools/androguard4/util.py diff --git a/LICENSES/androguard.txt b/LICENSES/androguard.txt index 0c2f45d5..49b27b8f 100644 --- a/LICENSES/androguard.txt +++ b/LICENSES/androguard.txt @@ -1,76 +1,178 @@ - Androguard +Androguard 4 -Copyright (C) 2012 - 2016, Anthony Desnos (desnos at t0t0.fr) All rights reserved. + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ -Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION - http://www.apache.org/licenses/LICENSE-2.0 + 1. Definitions. -Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. -================================================================================================================================================================================================================================================================================================================== + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. - DAD + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. -Copyright (C) 2012 - 2016, Geoffroy Gueguen (geoffroy dot gueguen at gmail dot com) All rights reserved. + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. -Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. - http://www.apache.org/licenses/LICENSE-2.0 + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. -Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). -================================================================================================================================================================================================================================================================================================================== -Apache License + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. -Version 2.0, January 2004 + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." -http://www.apache.org/licenses/ + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. -TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. -1. Definitions. + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. -"License" shall mean the terms and conditions for use, reproduction, and distribution as defined by Sections 1 through 9 of this document. + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: -"Licensor" shall mean the copyright owner or entity authorized by the copyright owner that is granting the License. + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and -"Legal Entity" shall mean the union of the acting entity and all other entities that control, are controlled by, or are under common control with that entity. For the purposes of this definition, "control" means (i) the power, direct or indirect, to cause the direction or management of such entity, whether by contract or otherwise, or (ii) ownership of fifty percent (50%) or more of the outstanding shares, or (iii) beneficial ownership of such entity. + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and -"You" (or "Your") shall mean an individual or Legal Entity exercising permissions granted by this License. + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and -"Source" form shall mean the preferred form for making modifications, including but not limited to software source code, documentation source, and configuration files. + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. -"Object" form shall mean any form resulting from mechanical transformation or translation of a Source form, including but not limited to compiled object code, generated documentation, and conversions to other media types. + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. -"Work" shall mean the work of authorship, whether in Source or Object form, made available under the License, as indicated by a copyright notice that is included in or attached to the work (an example is provided in the Appendix below). + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. -"Derivative Works" shall mean any work, whether in Source or Object form, that is based on (or derived from) the Work and for which the editorial revisions, annotations, elaborations, or other modifications represent, as a whole, an original work of authorship. For the purposes of this License, Derivative Works shall not include works that remain separable from, or merely link (or bind by name) to the interfaces of, the Work and Derivative Works thereof. + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. -"Contribution" shall mean any work of authorship, including the original version of the Work and any modifications or additions to that Work or Derivative Works thereof, that is intentionally submitted to Licensor for inclusion in the Work by the copyright owner or by an individual or Legal Entity authorized to submit on behalf of the copyright owner. For the purposes of this definition, "submitted" means any form of electronic, verbal, or written communication sent to the Licensor or its representatives, including but not limited to communication on electronic mailing lists, source code control systems, and issue tracking systems that are managed by, or on behalf of, the Licensor for the purpose of discussing and improving the Work, but excluding communication that is conspicuously marked or otherwise designated in writing by the copyright owner as "Not a Contribution." + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. -"Contributor" shall mean Licensor and any individual or Legal Entity on behalf of whom a Contribution has been received by Licensor and subsequently incorporated within the Work. + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. -2. Grant of Copyright License. Subject to the terms and conditions of this License, each Contributor hereby grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free, irrevocable copyright license to reproduce, prepare Derivative Works of, publicly display, publicly perform, sublicense, and distribute the Work and such Derivative Works in Source or Object form. - -3. Grant of Patent License. Subject to the terms and conditions of this License, each Contributor hereby grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free, irrevocable (except as stated in this section) patent license to make, have made, use, offer to sell, sell, import, and otherwise transfer the Work, where such license applies only to those patent claims licensable by such Contributor that are necessarily infringed by their Contribution(s) alone or by combination of their Contribution(s) with the Work to which such Contribution(s) was submitted. If You institute patent litigation against any entity (including a cross-claim or counterclaim in a lawsuit) alleging that the Work or a Contribution incorporated within the Work constitutes direct or contributory patent infringement, then any patent licenses granted to You under this License for that Work shall terminate as of the date such litigation is filed. - -4. Redistribution. You may reproduce and distribute copies of the Work or Derivative Works thereof in any medium, with or without modifications, and in Source or Object form, provided that You meet the following conditions: - - You must give any other recipients of the Work or Derivative Works a copy of this License; and - You must cause any modified files to carry prominent notices stating that You changed the files; and - You must retain, in the Source form of any Derivative Works that You distribute, all copyright, patent, trademark, and attribution notices from the Source form of the Work, excluding those notices that do not pertain to any part of the Derivative Works; and - If the Work includes a "NOTICE" text file as part of its distribution, then any Derivative Works that You distribute must include a readable copy of the attribution notices contained within such NOTICE file, excluding those notices that do not pertain to any part of the Derivative Works, in at least one of the following places: within a NOTICE text file distributed as part of the Derivative Works; within the Source form or documentation, if provided along with the Derivative Works; or, within a display generated by the Derivative Works, if and wherever such third-party notices normally appear. The contents of the NOTICE file are for informational purposes only and do not modify the License. You may add Your own attribution notices within Derivative Works that You distribute, alongside or as an addendum to the NOTICE text from the Work, provided that such additional attribution notices cannot be construed as modifying the License. - - You may add Your own copyright statement to Your modifications and may provide additional or different license terms and conditions for use, reproduction, or distribution of Your modifications, or for any such Derivative Works as a whole, provided Your use, reproduction, and distribution of the Work otherwise complies with the conditions stated in this License. - -5. Submission of Contributions. Unless You explicitly state otherwise, any Contribution intentionally submitted for inclusion in the Work by You to the Licensor shall be under the terms and conditions of this License, without any additional terms or conditions. Notwithstanding the above, nothing herein shall supersede or modify the terms of any separate license agreement you may have executed with Licensor regarding such Contributions. - -6. Trademarks. This License does not grant permission to use the trade names, trademarks, service marks, or product names of the Licensor, except as required for reasonable and customary use in describing the origin of the Work and reproducing the content of the NOTICE file. - -7. Disclaimer of Warranty. Unless required by applicable law or agreed to in writing, Licensor provides the Work (and each Contributor provides its Contributions) on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied, including, without limitation, any warranties or conditions of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A PARTICULAR PURPOSE. You are solely responsible for determining the appropriateness of using or redistributing the Work and assume any risks associated with Your exercise of permissions under this License. - -8. Limitation of Liability. In no event and under no legal theory, whether in tort (including negligence), contract, or otherwise, unless required by applicable law (such as deliberate and grossly negligent acts) or agreed to in writing, shall any Contributor be liable to You for damages, including any direct, indirect, special, incidental, or consequential damages of any character arising as a result of this License or out of the use or inability to use the Work (including but not limited to damages for loss of goodwill, work stoppage, computer failure or malfunction, or any and all other commercial damages or losses), even if such Contributor has been advised of the possibility of such damages. - -9. Accepting Warranty or Additional Liability. While redistributing the Work or Derivative Works thereof, You may choose to offer, and charge a fee for, acceptance of support, warranty, indemnity, or other liability obligations and/or rights consistent with this License. However, in accepting such obligations, You may act only on Your own behalf and on Your sole responsibility, not on behalf of any other Contributor, and only if You agree to indemnify, defend, and hold each Contributor harmless for any liability incurred by, or claims asserted against, such Contributor by reason of your accepting any such warranty or additional liability. + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + END OF TERMS AND CONDITIONS diff --git a/mobsf/MobSF/init.py b/mobsf/MobSF/init.py index 123114e8..14b6d8ab 100644 --- a/mobsf/MobSF/init.py +++ b/mobsf/MobSF/init.py @@ -18,7 +18,7 @@ from mobsf.install.windows.setup import windows_config_local logger = logging.getLogger(__name__) -VERSION = '4.2.6' +VERSION = '4.2.7' BANNER = r""" __ __ _ ____ _____ _ _ ____ | \/ | ___ | |__/ ___|| ___|_ _| || | |___ \ diff --git a/mobsf/StaticAnalyzer/tools/androguard4/apk.py b/mobsf/StaticAnalyzer/tools/androguard4/apk.py index a76ae648..6aac025f 100644 --- a/mobsf/StaticAnalyzer/tools/androguard4/apk.py +++ b/mobsf/StaticAnalyzer/tools/androguard4/apk.py @@ -1,89 +1,95 @@ # -*- coding: utf_8 -*- # flake8: noqa -# Androguard - -from .axml import ARSCParser, AXMLPrinter, ARSCResTableConfig -from .zipfile import ZipEntry +# Androguard4 APK - Nov 24, 2024 - 04a5703b8ba7c181bb9f5f5995a2c16b6f9353cf +# Allows type hinting of types not-yet-declared +# in Python >= 3.7 +# see https://peps.python.org/pep-0563/ +from __future__ import annotations # Python core +import binascii +import hashlib import io -from zlib import crc32 import os import re -import binascii +import unicodedata import zipfile +from hashlib import md5, sha1, sha224, sha256, sha384, sha512 from struct import unpack -import hashlib -import asn1crypto +from typing import Any, Iterator, List, Tuple, Union +from xml.dom.pulldom import SAX2DOM +from zlib import crc32 + +import lxml.sax +from .apkinspector.headers import ZipEntry + +# Used for reading Certificates +from asn1crypto import cms, keys, x509 +from asn1crypto.util import OrderedDict +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import dsa, ec, padding, rsa + + +# External dependencies +from lxml.etree import Element + +# Androguard +from .axml import ( + END_DOCUMENT, + END_TAG, + START_TAG, + TEXT, + ARSCParser, + ARSCResTableConfig, + AXMLParser, + AXMLPrinter, + format_value, +) +from .util import get_certificate_name_string + import logging -# External dependecies -import lxml.sax -from xml.dom.pulldom import SAX2DOM -# Used for reading Certificates -from asn1crypto import cms, x509, keys logger = logging.getLogger(__name__) logger.setLevel(level=logging.CRITICAL) - NS_ANDROID_URI = 'http://schemas.android.com/apk/res/android' NS_ANDROID = '{{{}}}'.format(NS_ANDROID_URI) # Namespace as used by etree - -def get_certificate_name_string(name, short=False, delimiter=', '): - """ - Function from androguard. - - licensed under the Apache License, Version 2.0. - https://github.com/androguard/androguard/blob/master/androguard/util.py - Format the Name type of a X509 Certificate in a human readable form. - - :param name: Name object to return the DN from - :param short: Use short form (default: False) - :param delimiter: Delimiter string or character between - two parts (default: ', ') - - :type name: dict or :class:`asn1crypto.x509.Name` - :type short: boolean - :type delimiter: str - - :rtype: str - """ - if isinstance(name, asn1crypto.x509.Name): - name = name.native - - # For the shortform, we have a lookup table - # See RFC4514 for more details - _ = { - 'business_category': ('businessCategory', 'businessCategory'), - 'serial_number': ('serialNumber', 'serialNumber'), - 'country_name': ('C', 'countryName'), - 'postal_code': ('postalCode', 'postalCode'), - 'state_or_province_name': ('ST', 'stateOrProvinceName'), - 'locality_name': ('L', 'localityName'), - 'street_address': ('street', 'streetAddress'), - 'organization_name': ('O', 'organizationName'), - 'organizational_unit_name': ('OU', 'organizationalUnitName'), - 'title': ('title', 'title'), - 'common_name': ('CN', 'commonName'), - 'initials': ('initials', 'initials'), - 'generation_qualifier': ('generationQualifier', 'generationQualifier'), - 'surname': ('SN', 'surname'), - 'given_name': ('GN', 'givenName'), - 'name': ('name', 'name'), - 'pseudonym': ('pseudonym', 'pseudonym'), - 'dn_qualifier': ('dnQualifier', 'dnQualifier'), - 'telephone_number': ('telephoneNumber', 'telephoneNumber'), - 'email_address': ('E', 'emailAddress'), - 'domain_component': ('DC', 'domainComponent'), - 'name_distinguisher': ('nameDistinguisher', 'nameDistinguisher'), - 'organization_identifier': ( - 'organizationIdentifier', 'organizationIdentifier'), - } - return delimiter.join( - ['{}={}'.format( - _.get(attr, (attr, attr))[0 if short else 1], - name[attr]) for attr in name]) +# Dictionary of the different protection levels mapped to their corresponding attribute names as described in +# https://android.googlesource.com/platform/frameworks/base/+/master/core/java/android/content/pm/PermissionInfo.java +protection_flags_to_attributes = { + "0x00000000": "normal", + "0x00000001": "dangerous", + "0x00000002": "signature", + "0x00000003": "signature or system", + "0x00000004": "internal", + "0x00000010": "privileged", + "0x00000020": "development", + "0x00000040": "appop", + "0x00000080": "pre23", + "0x00000100": "installer", + "0x00000200": "verifier", + "0x00000400": "preinstalled", + "0x00000800": "setup", + "0x00001000": "instant", + "0x00002000": "runtime only", + "0x00004000": "oem", + "0x00008000": "vendor privileged", + "0x00010000": "system text classifier", + "0x00020000": "wellbeing", + "0x00040000": "documenter", + "0x00080000": "configurator", + "0x00100000": "incident report approver", + "0x00200000": "app predictor", + "0x00400000": "module", + "0x00800000": "companion", + "0x01000000": "retail demo", + "0x02000000": "recents", + "0x04000000": "role", + "0x08000000": "known signer", +} def parse_lxml_dom(tree): @@ -94,6 +100,7 @@ def parse_lxml_dom(tree): class Error(Exception): """Base class for exceptions in this module.""" + pass @@ -106,7 +113,7 @@ class BrokenAPKError(Error): def _dump_additional_attributes(additional_attributes): - """ try to parse additional attributes, but ends up to hexdump if the scheme is unknown """ + """try to parse additional attributes, but ends up to hexdump if the scheme is unknown""" attributes_raw = io.BytesIO(additional_attributes) attributes_hex = binascii.hexlify(additional_attributes) @@ -114,15 +121,15 @@ def _dump_additional_attributes(additional_attributes): if not len(additional_attributes): return attributes_hex - len_attribute, = unpack(' None: self._bytes = None self.digests = None self.certificates = None @@ -163,21 +171,38 @@ class APKV2SignedData: certs_infos += "\n" certs_infos += " [%d]\n" % i certs_infos += " - Issuer: %s\n" % get_certificate_name_string( - x509_cert.issuer, short=True) + x509_cert.issuer, short=True + ) certs_infos += " - Subject: %s\n" % get_certificate_name_string( - x509_cert.subject, short=True) - certs_infos += " - Serial Number: %s\n" % hex(x509_cert.serial_number) + x509_cert.subject, short=True + ) + certs_infos += " - Serial Number: %s\n" % hex( + x509_cert.serial_number + ) certs_infos += " - Hash Algorithm: %s\n" % x509_cert.hash_algo - certs_infos += " - Signature Algorithm: %s\n" % x509_cert.signature_algo - certs_infos += " - Valid not before: %s\n" % x509_cert['tbs_certificate']['validity']['not_before'].native - certs_infos += " - Valid not after: %s" % x509_cert['tbs_certificate']['validity']['not_after'].native + certs_infos += ( + " - Signature Algorithm: %s\n" % x509_cert.signature_algo + ) + certs_infos += ( + " - Valid not before: %s\n" + % x509_cert['tbs_certificate']['validity']['not_before'].native + ) + certs_infos += ( + " - Valid not after: %s" + % x509_cert['tbs_certificate']['validity']['not_after'].native + ) - return "\n".join([ - 'additional_attributes : {}'.format( - _dump_additional_attributes(self.additional_attributes)), - 'digests : {}'.format(_dump_digests_or_signatures(self.digests)), - 'certificates : {}'.format(certs_infos), - ]) + return "\n".join( + [ + 'additional_attributes : {}'.format( + _dump_additional_attributes(self.additional_attributes) + ), + 'digests : {}'.format( + _dump_digests_or_signatures(self.digests) + ), + 'certificates : {}'.format(certs_infos), + ] + ) class APKV3SignedData(APKV2SignedData): @@ -186,7 +211,7 @@ class APKV3SignedData(APKV2SignedData): source : https://source.android.com/security/apksigning/v3.html """ - def __init__(self): + def __init__(self) -> None: super().__init__() self.minSDK = None self.maxSDK = None @@ -197,14 +222,16 @@ class APKV3SignedData(APKV2SignedData): # maxSDK is set to a negative value if there is no upper bound on the sdk targeted max_sdk_str = "%d" % self.maxSDK - if self.maxSDK >= 0x7fffffff: + if self.maxSDK >= 0x7FFFFFFF: max_sdk_str = "0x%x" % self.maxSDK - return "\n".join([ - 'signer minSDK : {:d}'.format(self.minSDK), - 'signer maxSDK : {:s}'.format(max_sdk_str), - base_str - ]) + return "\n".join( + [ + 'signer minSDK : {:d}'.format(self.minSDK), + 'signer maxSDK : {:s}'.format(max_sdk_str), + base_str, + ] + ) class APKV2Signer: @@ -213,18 +240,22 @@ class APKV2Signer: source : https://source.android.com/security/apksigning/v2.html """ - def __init__(self): + def __init__(self) -> None: self._bytes = None self.signed_data = None self.signatures = None self.public_key = None def __str__(self): - return "\n".join([ - '{:s}'.format(str(self.signed_data)), - 'signatures : {}'.format(_dump_digests_or_signatures(self.signatures)), - 'public key : {}'.format(binascii.hexlify(self.public_key)), - ]) + return "\n".join( + [ + '{:s}'.format(str(self.signed_data)), + 'signatures : {}'.format( + _dump_digests_or_signatures(self.signatures) + ), + 'public key : {}'.format(binascii.hexlify(self.public_key)), + ] + ) class APKV3Signer(APKV2Signer): @@ -233,7 +264,7 @@ class APKV3Signer(APKV2Signer): source : https://source.android.com/security/apksigning/v3.html """ - def __init__(self): + def __init__(self) -> None: super().__init__() self.minSDK = None self.maxSDK = None @@ -244,14 +275,16 @@ class APKV3Signer(APKV2Signer): # maxSDK is set to a negative value if there is no upper bound on the sdk targeted max_sdk_str = "%d" % self.maxSDK - if self.maxSDK >= 0x7fffffff: + if self.maxSDK >= 0x7FFFFFFF: max_sdk_str = "0x%x" % self.maxSDK - return "\n".join([ - 'signer minSDK : {:d}'.format(self.minSDK), - 'signer maxSDK : {:s}'.format(max_sdk_str), - base_str - ]) + return "\n".join( + [ + 'signer minSDK : {:d}'.format(self.minSDK), + 'signer maxSDK : {:s}'.format(max_sdk_str), + base_str, + ] + ) class APK: @@ -261,9 +294,9 @@ class APK: # Constants in the APK Signature Block _APK_SIG_MAGIC = b"APK Sig Block 42" - _APK_SIG_KEY_V2_SIGNATURE = 0x7109871a - _APK_SIG_KEY_V3_SIGNATURE = 0xf05368c0 - _APK_SIG_ATTR_V2_STRIPPING_PROTECTION = 0xbeeff00d + _APK_SIG_KEY_V2_SIGNATURE = 0x7109871A + _APK_SIG_KEY_V3_SIGNATURE = 0xF05368C0 + _APK_SIG_ATTR_V2_STRIPPING_PROTECTION = 0xBEEFF00D _APK_SIG_ALGO_IDS = { 0x0101: "RSASSA-PSS with SHA2-256 digest, SHA2-256 MGF1, 32 bytes of salt, trailer: 0xbc", @@ -279,7 +312,14 @@ class APK: __no_magic = False - def __init__(self, filename, raw=False, magic_file=None, skip_analysis=False, testzip=False): + def __init__( + self, + filename: str, + raw: bool = False, + magic_file: Union[str, None] = None, + skip_analysis: bool = False, + testzip: bool = False, + ) -> None: """ This class can access to all elements in an APK file @@ -303,7 +343,8 @@ class APK: """ if magic_file: logger.warning( - "You set magic_file but this parameter is actually unused. You should remove it.") + "You set magic_file but this parameter is actually unused. You should remove it." + ) self.filename = filename @@ -338,7 +379,9 @@ class APK: self.__raw = self.zip.zip.getvalue() if testzip: - logger.info("Testing zip file integrity, this might take a while...") + logger.info( + "Testing zip file integrity, this might take a while..." + ) # Test the zipfile for integrity before continuing. # This process might be slow, as the whole file is read. # Therefore it is possible to enable it as a separate feature. @@ -353,7 +396,8 @@ class APK: # That the filename is either very very long or does not make any sense. # Thus we do not do it, the user might find out by using other tools. raise BrokenAPKError( - "The APK is probably broken: testzip returned an error.") + "The APK is probably broken: testzip returned an error." + ) if not skip_analysis: self._apk_analysis() @@ -384,7 +428,8 @@ class APK: if not ap.is_valid(): logger.error( - "Error while parsing AndroidManifest.xml - is the file valid?") + "Error while parsing AndroidManifest.xml - is the file valid?" + ) return self.axml[i] = ap @@ -392,41 +437,63 @@ class APK: if self.axml[i].is_packed(): logger.warning( - "XML Seems to be packed, operations on the AndroidManifest.xml might fail.") + "XML Seems to be packed, operations on the AndroidManifest.xml might fail." + ) if self.xml[i] is not None: if self.xml[i].tag != "manifest": logger.error( - "AndroidManifest.xml does not start with a tag! Is this a valid APK?") + "AndroidManifest.xml does not start with a tag! Is this a valid APK?" + ) return self.package = self.get_attribute_value("manifest", "package") self.androidversion["Code"] = self.get_attribute_value( - "manifest", "versionCode") + "manifest", "versionCode" + ) self.androidversion["Name"] = self.get_attribute_value( - "manifest", "versionName") - permission = list(self.get_all_attribute_value( - "uses-permission", "name")) + "manifest", "versionName" + ) + permission = list( + self.get_all_attribute_value("uses-permission", "name") + ) self.permissions = list(set(self.permissions + permission)) for uses_permission in self.find_tags("uses-permission"): - self.uses_permissions.append([ - self.get_value_from_tag(uses_permission, "name"), - self._get_permission_maxsdk(uses_permission) - ]) + self.uses_permissions.append( + [ + self.get_value_from_tag(uses_permission, "name"), + self._get_permission_maxsdk(uses_permission), + ] + ) # getting details of the declared permissions for d_perm_item in self.find_tags('permission'): d_perm_name = self._get_res_string_value( - str(self.get_value_from_tag(d_perm_item, "name"))) + str(self.get_value_from_tag(d_perm_item, "name")) + ) d_perm_label = self._get_res_string_value( - str(self.get_value_from_tag(d_perm_item, "label"))) + str(self.get_value_from_tag(d_perm_item, "label")) + ) d_perm_description = self._get_res_string_value( - str(self.get_value_from_tag(d_perm_item, "description"))) + str( + self.get_value_from_tag(d_perm_item, "description") + ) + ) d_perm_permissionGroup = self._get_res_string_value( - str(self.get_value_from_tag(d_perm_item, "permissionGroup"))) + str( + self.get_value_from_tag( + d_perm_item, "permissionGroup" + ) + ) + ) d_perm_protectionLevel = self._get_res_string_value( - str(self.get_value_from_tag(d_perm_item, "protectionLevel"))) + str( + self.get_value_from_tag( + d_perm_item, "protectionLevel" + ) + ) + ) d_perm_details = { "label": d_perm_label, @@ -437,12 +504,16 @@ class APK: self.declared_permissions[d_perm_name] = d_perm_details self.valid_apk = True - logger.debug("APK file was successfully validated!") + logger.info("APK file was successfully validated!") # self.permission_module = androconf.load_api_specific_resource_module( - # "aosp_permissions", self.get_target_sdk_version()) - # self.permission_module_min_sdk = androconf.load_api_specific_resource_module( - # "aosp_permissions", self.get_min_sdk_version()) + # "aosp_permissions", self.get_target_sdk_version() + # ) + # self.permission_module_min_sdk = ( + # androconf.load_api_specific_resource_module( + # "aosp_permissions", self.get_min_sdk_version() + # ) + # ) def __getstate__(self): """ @@ -494,13 +565,15 @@ class APK: try: maxSdkVersion = int(self.get_value_from_tag(item, "maxSdkVersion")) except ValueError: - logger.warning(str(maxSdkVersion) - + ' is not a valid value for maxSdkVersion') + logger.warning( + str(maxSdkVersion) + + ' is not a valid value for maxSdkVersion' + ) except TypeError: pass return maxSdkVersion - def is_valid_APK(self): + def is_valid_APK(self) -> bool: """ Return true if the APK is valid, false otherwise. An APK is seen as valid, if the AndroidManifest.xml could be successful parsed. @@ -511,7 +584,7 @@ class APK: """ return self.valid_apk - def get_filename(self): + def get_filename(self) -> str: """ Return the filename of the APK @@ -519,7 +592,7 @@ class APK: """ return self.filename - def get_app_name(self): + def get_app_name(self, locale=None) -> str: """ Return the appname of the APK @@ -542,13 +615,15 @@ class APK: # FIXME: would need to use _format_value inside get_attribute_value for each returned name! # For example, as the activity name might be foobar.foo.bar but inside the activity it is only .bar app_name = self.get_attribute_value( - 'activity', 'label', name=main_activity_name) + 'activity', 'label', name=main_activity_name + ) if app_name is None: # No App name set # TODO return packagename instead? logger.warning( - "It looks like that no app name is set for the main activity!") + "It looks like that no app name is set for the main activity!" + ) return "" if app_name.startswith("@"): @@ -565,24 +640,34 @@ class APK: if package == 'android': # TODO: we can not resolve this, as we lack framework-res.apk # one exception would be when parsing framework-res.apk directly. - logger.warning("Resource ID with android package name encountered! " - "Will not resolve, framework-res.apk would be required.") + logger.warning( + "Resource ID with android package name encountered! " + "Will not resolve, framework-res.apk would be required." + ) return app_name else: # TODO should look this up, might be in the resources logger.warning( - "Resource ID with Package name '{}' encountered! Will not resolve".format(package)) + "Resource ID with Package name '{}' encountered! Will not resolve".format( + package + ) + ) return app_name try: - app_name = res_parser.get_resolved_res_configs( - res_id, - ARSCResTableConfig.default_config())[0][1] + config = ( + ARSCResTableConfig(None, locale=locale) + if locale + else ARSCResTableConfig.default_config() + ) + app_name = res_parser.get_resolved_res_configs(res_id, config)[ + 0 + ][1] except Exception as e: logger.warning("Exception selecting app name: %s" % e) return app_name - def get_app_icon(self, max_dpi=65536): + def get_app_icon(self, max_dpi: int = 65536) -> Union[str, None]: """ Return the first icon file name, which density is not greater than max_dpi, unless exact icon resolution is set in the manifest, in which case @@ -626,7 +711,8 @@ class APK: main_activity_name = self.get_main_activity() app_icon = self.get_attribute_value( - 'activity', 'icon', name=main_activity_name) + 'activity', 'icon', name=main_activity_name + ) if not app_icon: app_icon = self.get_attribute_value('application', 'icon') @@ -637,13 +723,16 @@ class APK: return None if not app_icon: - res_id = res_parser.get_res_id_by_key(self.package, 'mipmap', 'ic_launcher') + res_id = res_parser.get_res_id_by_key( + self.package, 'mipmap', 'ic_launcher' + ) if res_id: app_icon = "@%x" % res_id if not app_icon: res_id = res_parser.get_res_id_by_key( - self.package, 'drawable', 'ic_launcher') + self.package, 'drawable', 'ic_launcher' + ) if res_id: app_icon = "@%x" % res_id @@ -671,7 +760,7 @@ class APK: return app_icon - def get_package(self): + def get_package(self) -> str: """ Return the name of the package @@ -681,7 +770,7 @@ class APK: """ return self.package - def get_androidversion_code(self): + def get_androidversion_code(self) -> str: """ Return the android version code @@ -691,7 +780,7 @@ class APK: """ return self.androidversion["Code"] - def get_androidversion_name(self): + def get_androidversion_name(self) -> str: """ Return the android version name @@ -701,7 +790,7 @@ class APK: """ return self.androidversion["Name"] - def get_files(self): + def get_files(self) -> list[str]: """ Return the file names inside the APK. @@ -709,7 +798,7 @@ class APK: """ return self.zip.namelist() - # def _get_file_magic_name(self, buffer): + # def _get_file_magic_name(self, buffer: bytes) -> str: # """ # Return the filetype guessed for a buffer # :param buffer: bytes @@ -730,52 +819,60 @@ class APK: # return default # except TypeError as e: # self.__no_magic = True - # logger.warning("It looks like you have the magic python package installed but not the magic library itself!") + # logger.warning( + # "It looks like you have the magic python package installed but not the magic library itself!" + # ) # logger.warning("Error from magic library: %s", e) - # logger.warning("Please follow the installation instructions at https://github.com/ahupp/python-magic/#installation") - # logger.warning("You can also install the 'python-magic-bin' package on Windows and MacOS") + # logger.warning( + # "Please follow the installation instructions at https://github.com/ahupp/python-magic/#installation" + # ) + # logger.warning( + # "You can also install the 'python-magic-bin' package on Windows and MacOS" + # ) # return default - try: - # There are several implementations of magic, - # unfortunately all called magic - # We use this one: https://github.com/ahupp/python-magic/ - # You can also use python-magic-bin on Windows or MacOS - getattr(magic, "MagicException") - except AttributeError: - self.__no_magic = True - logger.warning("Not the correct Magic library was found on your " - "system. Please install python-magic or python-magic-bin!") - return default + # try: + # # There are several implementations of magic, + # # unfortunately all called magic + # # We use this one: https://github.com/ahupp/python-magic/ + # # You can also use python-magic-bin on Windows or MacOS + # getattr(magic, "MagicException") + # except AttributeError: + # self.__no_magic = True + # logger.warning( + # "Not the correct Magic library was found on your " + # "system. Please install python-magic or python-magic-bin!" + # ) + # return default - try: - # 1024 byte are usually enough to test the magic - ftype = magic.from_buffer(buffer[:1024]) - except magic.MagicException as e: - logger.exception("Error getting the magic type: %s", e) - return default + # try: + # # 1024 byte are usually enough to test the magic + # ftype = magic.from_buffer(buffer[:1024]) + # except magic.MagicException as e: + # logger.exception("Error getting the magic type: %s", e) + # return default - if not ftype: - return default - else: - return self._patch_magic(buffer, ftype) + # if not ftype: + # return default + # else: + # return self._patch_magic(buffer, ftype) - @property - def files(self): - """ - Returns a dictionary of filenames and detected magic type + # @property + # def files(self) -> dict[str, str]: + # """ + # Returns a dictionary of filenames and detected magic type - :returns: dictionary of files and their mime type - """ - return self.get_files_types() + # :returns: dictionary of files and their mime type + # """ + # return self.get_files_types() - # def get_files_types(self): + # def get_files_types(self) -> dict[str, str]: # """ # Return the files inside the APK with their associated types (by using python-magic) # At the same time, the CRC32 are calculated for the files. - # :rtype: a dictionnary + # :rtype: a dictionary # """ # if self._files == {}: # # Generate File Types / CRC List @@ -793,7 +890,11 @@ class APK: # :param orig: guess by mime libary # :returns: corrected guess # """ - # if ("Zip" in orig) or ('(JAR)' in orig) and androconf.is_android_raw(buffer) == 'APK': + # if ( + # ("Zip" in orig) + # or ('(JAR)' in orig) + # and androconf.is_android_raw(buffer) == 'APK' + # ): # return "Android application package file" # return orig @@ -810,15 +911,23 @@ class APK: buffer = self.zip.read(filename) if filename not in self.files_crc32: self.files_crc32[filename] = crc32(buffer) - if self.files_crc32[filename] != self.zip.infolist()[filename].crc32_of_uncompressed_data: - logger.error("File '{}' has different CRC32 after unpacking! " - "Declared: {:08x}, Calculated: {:08x}".format(filename, - self.zip.infolist()[ - filename].crc32_of_uncompressed_data, - self.files_crc32[filename])) + if ( + self.files_crc32[filename] + != self.zip.infolist()[filename].crc32_of_uncompressed_data + ): + logger.error( + "File '{}' has different CRC32 after unpacking! " + "Declared: {:08x}, Calculated: {:08x}".format( + filename, + self.zip.infolist()[ + filename + ].crc32_of_uncompressed_data, + self.files_crc32[filename], + ) + ) return buffer - def get_files_crc32(self): + def get_files_crc32(self) -> dict[str, int]: """ Calculates and returns a dictionary of filenames and CRC32 @@ -830,16 +939,16 @@ class APK: return self.files_crc32 - # def get_files_information(self): - # """ - # Return the files inside the APK with their associated types and crc32 + def get_files_information(self) -> Iterator[tuple[str, str, int]]: + """ + Return the files inside the APK with their associated types and crc32 - # :rtype: str, str, int - # """ - # for k in self.get_files(): - # yield k, self.get_files_types()[k], self.get_files_crc32()[k] + :rtype: str, str, int + """ + for k in self.get_files(): + yield k, self.get_files_types()[k], self.get_files_crc32()[k] - def get_raw(self): + def get_raw(self) -> bytes: """ Return raw bytes of the APK @@ -853,7 +962,7 @@ class APK: self.__raw = bytearray(f.read()) return self.__raw - def get_file(self, filename): + def get_file(self, filename: str) -> bytes: """ Return the raw data of the specified filename inside the APK @@ -865,7 +974,7 @@ class APK: except KeyError: raise FileNotPresent(filename) - def get_dex(self): + def get_dex(self) -> bytes: """ Return the raw data of the classes dex file @@ -880,7 +989,7 @@ class APK: # TODO is this a good idea to return an empty string? return b"" - def get_dex_names(self): + def get_dex_names(self) -> list[str]: """ Return the names of all DEX files found in the APK. This method only accounts for "offical" dex files, i.e. all files @@ -891,7 +1000,7 @@ class APK: dexre = re.compile(r"^classes(\d*).dex$") return filter(lambda x: dexre.match(x), self.get_files()) - def get_all_dex(self): + def get_all_dex(self) -> Iterator[bytes]: """ Return the raw data of all classes dex files @@ -900,14 +1009,23 @@ class APK: for dex_name in self.get_dex_names(): yield self.get_file(dex_name) - def is_multidex(self): + def is_multidex(self) -> bool: """ Test if the APK has multiple DEX files :returns: True if multiple dex found, otherwise False """ dexre = re.compile(r"^classes(\d+)?.dex$") - return len([instance for instance in self.get_files() if dexre.search(instance)]) > 1 + return ( + len( + [ + instance + for instance in self.get_files() + if dexre.search(instance) + ] + ) + > 1 + ) def _format_value(self, value): """ @@ -931,8 +1049,12 @@ class APK: return value def get_all_attribute_value( - self, tag_name, attribute, format_value=True, **attribute_filter - ): + self, + tag_name: str, + attribute: str, + format_value: bool = True, + **attribute_filter, + ) -> Iterator[str]: """ Yields all the attribute values in xml files which match with the tag name and the specific attribute @@ -950,8 +1072,12 @@ class APK: yield value def get_attribute_value( - self, tag_name, attribute, format_value=False, **attribute_filter - ): + self, + tag_name: str, + attribute: str, + format_value: bool = False, + **attribute_filter, + ) -> str: """ Return the attribute value in xml files which matches the tag name and the specific attribute @@ -961,11 +1087,14 @@ class APK: """ for value in self.get_all_attribute_value( - tag_name, attribute, format_value, **attribute_filter): + tag_name, attribute, format_value, **attribute_filter + ): if value is not None: return value - def get_value_from_tag(self, tag, attribute): + def get_value_from_tag( + self, tag: Element, attribute: str + ) -> Union[str, None]: """ Return the value of the android prefixed attribute in a specific tag. @@ -1005,27 +1134,29 @@ class APK: if value: # If value is still None, the attribute could not be found, thus is not present - logger.warning("Failed to get the attribute '{}' on tag '{}' with namespace. " - "But found the same attribute without namespace!".format(attribute, tag.tag)) + logger.warning( + "Failed to get the attribute '{}' on tag '{}' with namespace. " + "But found the same attribute without namespace!".format( + attribute, tag.tag + ) + ) return value - def find_tags(self, tag_name, **attribute_filter): + def find_tags(self, tag_name: str, **attribute_filter) -> list[str]: """ Return a list of all the matched tags in all available xml :param str tag: specify the tag name """ all_tags = [ - self.find_tags_from_xml( - i, tag_name, **attribute_filter - ) + self.find_tags_from_xml(i, tag_name, **attribute_filter) for i in self.xml ] return [tag for tag_list in all_tags for tag in tag_list] def find_tags_from_xml( - self, xml_name, tag_name, **attribute_filter - ): + self, xml_name: str, tag_name: str, **attribute_filter + ) -> list[str]: """ Return a list of all the matched tags in a specific xml w @@ -1036,19 +1167,20 @@ class APK: if xml is None: return [] if xml.tag == tag_name: - if self.is_tag_matched( - xml.tag, **attribute_filter - ): + if self.is_tag_matched(xml.tag, **attribute_filter): return [xml] return [] - tags = xml.findall(".//" + tag_name) + tags = set() + tags.update(xml.findall(".//" + tag_name)) + + # https://github.com/androguard/androguard/pull/1053 + # permission declared using tag bool: r""" Return true if the attributes matches in attribute filter. @@ -1075,7 +1207,7 @@ class APK: return False return True - def get_main_activities(self): + def get_main_activities(self) -> set[str]: """ Return names of the main activities @@ -1089,8 +1221,9 @@ class APK: for i in self.xml: if self.xml[i] is None: continue - activities_and_aliases = self.xml[i].findall(".//activity") + \ - self.xml[i].findall(".//activity-alias") + activities_and_aliases = self.xml[i].findall( + ".//activity" + ) + self.xml[i].findall(".//activity-alias") for item in activities_and_aliases: # Some applications have more than one MAIN activity. @@ -1119,7 +1252,7 @@ class APK: return x.intersection(y) - def get_main_activity(self): + def get_main_activity(self) -> Union[str, None]: """ Return the name of the main activity @@ -1135,13 +1268,14 @@ class APK: # sorted is necessary # 9fc7d3e8225f6b377f9181a92c551814317b77e1aa0df4c6d508d24b18f0f633 good_main_activities = sorted( - main_activities.intersection(self.get_activities())) + main_activities.intersection(self.get_activities()) + ) if good_main_activities: return good_main_activities[0] return sorted(main_activities)[0] return None - def get_activities(self): + def get_activities(self) -> list[str]: """ Return the android:name attribute of all activities @@ -1149,7 +1283,7 @@ class APK: """ return list(self.get_all_attribute_value("activity", "name")) - def get_activity_aliases(self): + def get_activity_aliases(self) -> list[dict[str, str]]: """ Return the android:name and android:targetActivity attribute of all activity aliases. @@ -1159,8 +1293,7 @@ class APK: for alias in self.find_tags('activity-alias'): activity_alias = {} for attribute in ['name', 'targetActivity']: - value = (alias.get(attribute) - or alias.get(self._ns(attribute))) + value = alias.get(attribute) or alias.get(self._ns(attribute)) if not value: continue activity_alias[attribute] = self._format_value(value) @@ -1168,7 +1301,7 @@ class APK: ali.append(activity_alias) return ali - def get_services(self): + def get_services(self) -> list[str]: """ Return the android:name attribute of all services @@ -1176,7 +1309,7 @@ class APK: """ return list(self.get_all_attribute_value("service", "name")) - def get_receivers(self): + def get_receivers(self) -> list[str]: """ Return the android:name attribute of all receivers @@ -1184,7 +1317,7 @@ class APK: """ return list(self.get_all_attribute_value("receiver", "name")) - def get_providers(self): + def get_providers(self) -> list[str]: """ Return the android:name attribute of all providers @@ -1192,7 +1325,7 @@ class APK: """ return list(self.get_all_attribute_value("provider", "name")) - def get_res_value(self, name): + def get_res_value(self, name: str) -> str: """ Return the literal value with a resource id :rtype: str @@ -1205,15 +1338,17 @@ class APK: res_id = res_parser.parse_id(name)[0] try: value = res_parser.get_resolved_res_configs( - res_id, - ARSCResTableConfig.default_config())[0][1] + res_id, ARSCResTableConfig.default_config() + )[0][1] except Exception as e: logger.warning("Exception get resolved resource id: %s" % e) return name return value - def get_intent_filters(self, itemtype, name): + def get_intent_filters( + self, itemtype: str, name: str + ) -> dict[str, list[str]]: """ Find intent filters for a given item and name. @@ -1225,8 +1360,19 @@ class APK: :param name: the `android:name` of the parent item, e.g. activity name :returns: a dictionary with the keys `action` and `category` containing the `android:name` of those items """ - attributes = {"action": ["name"], "category": ["name"], "data": [ - 'scheme', 'host', 'port', 'path', 'pathPattern', 'pathPrefix', 'mimeType']} + attributes = { + "action": ["name"], + "category": ["name"], + "data": [ + 'scheme', + 'host', + 'port', + 'path', + 'pathPattern', + 'pathPrefix', + 'mimeType', + ], + } d = {} for element in attributes.keys(): @@ -1245,7 +1391,9 @@ class APK: value = ssitem.get(self._ns(attribute)) if value: if value.startswith('@'): - value = self.get_res_value(value) + value = self.get_res_value( + value + ) values[attribute] = value if values: @@ -1265,7 +1413,7 @@ class APK: return d - def get_permissions(self): + def get_permissions(self) -> list[str]: """ Return permissions names declared in the AndroidManifest.xml. @@ -1282,11 +1430,11 @@ class APK: """ return self.permissions - def get_uses_implied_permission_list(self): + def get_uses_implied_permission_list(self) -> list[str]: """ - Return all permissions implied by the target SDK or other permissions. + Return all permissions implied by the target SDK or other permissions. - :rtype: list of string + :rtype: list of string """ target_sdk_version = self.get_effective_target_sdk_version() @@ -1308,8 +1456,10 @@ class APK: if READ_PHONE_STATE not in self.permissions: implied.append([READ_PHONE_STATE, None]) - if (WRITE_EXTERNAL_STORAGE in self.permissions or implied_WRITE_EXTERNAL_STORAGE) \ - and READ_EXTERNAL_STORAGE not in self.permissions: + if ( + WRITE_EXTERNAL_STORAGE in self.permissions + or implied_WRITE_EXTERNAL_STORAGE + ) and READ_EXTERNAL_STORAGE not in self.permissions: maxSdkVersion = None for name, version in self.uses_permissions: if name == WRITE_EXTERNAL_STORAGE: @@ -1318,16 +1468,22 @@ class APK: implied.append([READ_EXTERNAL_STORAGE, maxSdkVersion]) if target_sdk_version < 16: - if READ_CONTACTS in self.permissions \ - and READ_CALL_LOG not in self.permissions: + if ( + READ_CONTACTS in self.permissions + and READ_CALL_LOG not in self.permissions + ): implied.append([READ_CALL_LOG, None]) - if WRITE_CONTACTS in self.permissions \ - and WRITE_CALL_LOG not in self.permissions: + if ( + WRITE_CONTACTS in self.permissions + and WRITE_CALL_LOG not in self.permissions + ): implied.append([WRITE_CALL_LOG, None]) return implied - def _update_permission_protection_level(self, protection_level, sdk_version): + def _update_permission_protection_level( + self, protection_level, sdk_version + ): if not sdk_version or int(sdk_version) <= 15: return protection_level.replace('Or', '|').lower() return protection_level @@ -1337,22 +1493,32 @@ class APK: target_sdk = self.get_target_sdk_version() filled_permissions = permissions.copy() for permission in filled_permissions: - protection_level, label, description = filled_permissions[permission] - if ((not label or not description) - and permission in self.permission_module_min_sdk): + protection_level, label, description = filled_permissions[ + permission + ] + if ( + not label or not description + ) and permission in self.permission_module_min_sdk: x = self.permission_module_min_sdk[permission] protection_level = self._update_permission_protection_level( - x['protectionLevel'], min_sdk) + x['protectionLevel'], min_sdk + ) filled_permissions[permission] = [ - protection_level, x['label'], x['description']] + protection_level, + x['label'], + x['description'], + ] else: filled_permissions[permission] = [ self._update_permission_protection_level( - protection_level, target_sdk), - label, description] + protection_level, target_sdk + ), + label, + description, + ] return filled_permissions - def get_details_permissions(self): + def get_details_permissions(self) -> dict[str, list[str]]: """ Return permissions with details. @@ -1367,13 +1533,24 @@ class APK: if i in self.permission_module: x = self.permission_module[i] l[i] = [x["protectionLevel"], x["label"], x["description"]] + elif i in self.declared_permissions: + protectionLevel_hex = self.declared_permissions[i][ + "protectionLevel" + ] + protectionLevel = protection_flags_to_attributes[ + protectionLevel_hex + ] + l[i] = [ + protectionLevel, + "Unknown permission from android reference", + "Unknown permission from android reference", + ] else: - # FIXME: the permission might be signature, if it is defined by the app itself! - l[i] = ["normal", "Unknown permission from android reference", - "Unknown permission from android reference"] + # Is there a valid case not belonging to the above two? + logger.info(f"Unknown permission {i}") return self._fill_deprecated_permissions(l) - def get_requested_aosp_permissions(self): + def get_requested_aosp_permissions(self) -> list[str]: """ Returns requested permissions declared within AOSP project. @@ -1388,7 +1565,7 @@ class APK: aosp_permissions.append(perm) return aosp_permissions - def get_requested_aosp_permissions_details(self): + def get_requested_aosp_permissions_details(self) -> dict[str, list[str]]: """ Returns requested aosp permissions with details. @@ -1403,7 +1580,7 @@ class APK: continue return l - def get_requested_third_party_permissions(self): + def get_requested_third_party_permissions(self) -> list[str]: """ Returns list of requested permissions not declared within AOSP project. @@ -1416,7 +1593,7 @@ class APK: third_party_permissions.append(perm) return third_party_permissions - def get_declared_permissions(self): + def get_declared_permissions(self) -> list[str]: """ Returns list of the declared permissions. @@ -1424,7 +1601,7 @@ class APK: """ return list(self.declared_permissions.keys()) - def get_declared_permissions_details(self): + def get_declared_permissions_details(self) -> dict[str, list[str]]: """ Returns declared permissions with the details. @@ -1432,39 +1609,39 @@ class APK: """ return self.declared_permissions - def get_max_sdk_version(self): + def get_max_sdk_version(self) -> str: """ - Return the android:maxSdkVersion attribute + Return the android:maxSdkVersion attribute - :rtype: string + :rtype: string """ return self.get_attribute_value("uses-sdk", "maxSdkVersion") - def get_min_sdk_version(self): + def get_min_sdk_version(self) -> str: """ - Return the android:minSdkVersion attribute + Return the android:minSdkVersion attribute - :rtype: string + :rtype: string """ return self.get_attribute_value("uses-sdk", "minSdkVersion") - def get_target_sdk_version(self): + def get_target_sdk_version(self) -> str: """ - Return the android:targetSdkVersion attribute + Return the android:targetSdkVersion attribute - :rtype: string + :rtype: string """ return self.get_attribute_value("uses-sdk", "targetSdkVersion") - def get_effective_target_sdk_version(self): + def get_effective_target_sdk_version(self) -> int: """ - Return the effective targetSdkVersion, always returns int > 0. + Return the effective targetSdkVersion, always returns int > 0. - If the targetSdkVersion is not set, it defaults to 1. This is - set based on defaults as defined in: - https://developer.android.com/guide/topics/manifest/uses-sdk-element.html + If the targetSdkVersion is not set, it defaults to 1. This is + set based on defaults as defined in: + https://developer.android.com/guide/topics/manifest/uses-sdk-element.html - :rtype: int + :rtype: int """ target_sdk_version = self.get_target_sdk_version() if not target_sdk_version: @@ -1474,15 +1651,15 @@ class APK: except (ValueError, TypeError): return 1 - def get_libraries(self): + def get_libraries(self) -> list[str]: """ - Return the android:name attributes for libraries + Return the android:name attributes for libraries - :rtype: list + :rtype: list """ return list(self.get_all_attribute_value("uses-library", "name")) - def get_features(self): + def get_features(self) -> list[str]: """ Return a list of all android:names found for the tag uses-feature in the AndroidManifest.xml @@ -1491,7 +1668,7 @@ class APK: """ return list(self.get_all_attribute_value("uses-feature", "name")) - def is_wearable(self): + def is_wearable(self) -> bool: """ Checks if this application is build for wearables by checking if it uses the feature 'android.hardware.type.watch' @@ -1504,7 +1681,7 @@ class APK: """ return 'android.hardware.type.watch' in self.get_features() - def is_leanback(self): + def is_leanback(self) -> bool: """ Checks if this application is build for TV (Leanback support) by checkin if it uses the feature 'android.software.leanback' @@ -1513,7 +1690,7 @@ class APK: """ return 'android.software.leanback' in self.get_features() - def is_androidtv(self): + def is_androidtv(self) -> bool: """ Checks if this application does not require a touchscreen, as this is the rule to get into the TV section of the Play Store @@ -1521,22 +1698,297 @@ class APK: :returns: True if 'android.hardware.touchscreen' is not required, False otherwise """ - return self.get_attribute_value('uses-feature', 'name', required="false", name="android.hardware.touchscreen") == "android.hardware.touchscreen" + return ( + self.get_attribute_value( + 'uses-feature', + 'name', + required="false", + name="android.hardware.touchscreen", + ) + == "android.hardware.touchscreen" + ) - def get_certificate_der(self, filename): + def get_certificate_der( + self, filename: str, max_sdk_version: int = None + ) -> Union[bytes, None]: """ Return the DER coded X.509 certificate from the signature file. + If minSdkVersion is prior to Android N only the first SignerInfo is used. + If signed attributes are present, they are taken into account + Note that unsupported critical extensions and key usage are not verified! + https://android.googlesource.com/platform/tools/apksig/+/refs/tags/platform-tools-34.0.5/src/main/java/com/android/apksig/internal/apk/v1/V1SchemeVerifier.java#668 :param filename: Signature filename in APK - :returns: DER coded X.509 certificate as binary + :param max_sdk_version: An optional integer parameter for the max sdk version + :returns: DER coded X.509 certificate as binary or None """ + + # Get the signature pkcs7message = self.get_file(filename) + # Get the .SF + sf_filename = os.path.splitext(filename)[0] + '.SF' + sf_object = self.get_file(sf_filename) + # Load the signature + signed_data = cms.ContentInfo.load(pkcs7message) + # Locate the SignerInfo structure + signer_infos = signed_data['content']['signer_infos'] + if not signer_infos: + logger.error( + 'No signer information found in the PKCS7 object. The APK may not be properly signed.' + ) + return None - pkcs7obj = cms.ContentInfo.load(pkcs7message) - cert = pkcs7obj['content']['certificates'][0].chosen.dump() - return cert + # Prior to Android N, Android attempts to verify only the first SignerInfo. From N onwards, Android attempts + # to verify all SignerInfos and then picks the first verified SignerInfo. + min_sdk_version = self.get_min_sdk_version() + if ( + min_sdk_version is None or int(min_sdk_version) < 24 + ): # AndroidSdkVersion.N + logger.info( + f"minSdkVersion: {min_sdk_version} is less than 24. Getting the first signerInfo only!" + ) + unverified_signer_infos_to_try = [signer_infos[0]] + else: + unverified_signer_infos_to_try = signer_infos - def get_certificate(self, filename): + # Extract certificates from the PKCS7 object + certificates = signed_data['content']['certificates'] + return_certificate = None + list_certificates_verified = [] + for signer_info in unverified_signer_infos_to_try: + try: + matching_certificate_verified = ( + self.verify_signer_info_against_sig_file( + signed_data, + certificates, + signer_info, + sf_object, + max_sdk_version, + ) + ) + except (ValueError, TypeError, OSError, InvalidSignature) as e: + logger.error( + f"The following exception was raised while verifying the certificate: {e}" + ) + return ( + None # the validation stops due to the exception raised! + ) + if matching_certificate_verified is not None: + list_certificates_verified.append( + matching_certificate_verified + ) + if not list_certificates_verified: + logger.error( + f"minSdkVersion: {min_sdk_version}, # of SignerInfos: {len(unverified_signer_infos_to_try)}. None Verified!" + ) + else: + return_certificate = list_certificates_verified[0] + return return_certificate + + def verify_signer_info_against_sig_file( + self, + signed_data, + certificates, + signer_info, + sf_object, + max_sdk_version, + ): + matching_certificate = self.find_certificate(certificates, signer_info) + matching_certificate_verified = None + digest_algorithm, crypto_hash_algorithm = self.get_hash_algorithm( + signer_info + ) + if matching_certificate is None: + raise ValueError( + "Signing certificate referenced in SignerInfo not found in SignedData" + ) + else: + if signer_info['signed_attrs'].native: + logger.info("Signed Attributes detected!") + signed_attrs = signer_info['signed_attrs'] + signed_attrs_dict = OrderedDict() + for attr in signed_attrs: + if attr['type'].dotted in signed_attrs_dict: + raise ValueError( + f"Duplicate signed attribute: {attr['type'].dotted}" + ) + signed_attrs_dict[attr['type'].dotted] = attr['values'] + + # Check content type attribute (for Android N and newer) + if max_sdk_version is None or int(max_sdk_version) >= 24: + content_type_oid = ( + '1.2.840.113549.1.9.3' # OID for contentType + ) + if content_type_oid not in signed_attrs_dict: + raise ValueError( + "No Content Type in signed attributes" + ) + content_type = signed_attrs_dict[content_type_oid][ + 0 + ].native + if ( + content_type + != signed_data['content']['encap_content_info'][ + 'content_type' + ].native + ): + logger.error( + "Content Type mismatch. Continuing to next SignerInfo, if any." + ) + return None + + # Check message digest attribute + message_digest_oid = ( + '1.2.840.113549.1.9.4' # OID for messageDigest + ) + if message_digest_oid not in signed_attrs_dict: + raise ValueError("No content digest in signed attributes") + expected_signature_file_digest = signed_attrs_dict[ + message_digest_oid + ][0].native + hash_algo = digest_algorithm() + hash_algo.update(sf_object) + actual_digest = hash_algo.digest() + + # Compare digests + if actual_digest != expected_signature_file_digest: + logger.error( + "Digest mismatch. Continuing to next SignerInfo, if any." + ) + return None + + signed_attrs_dump = signed_attrs.dump() + # Modify the first byte to 0x31 for UNIVERSAL SET + signed_attrs_dump = b'\x31' + signed_attrs_dump[1:] + matching_certificate_verified = self.verify_signature( + signer_info, + matching_certificate, + signed_attrs_dump, + crypto_hash_algorithm, + ) + else: + matching_certificate_verified = self.verify_signature( + signer_info, + matching_certificate, + sf_object, + crypto_hash_algorithm, + ) + return matching_certificate_verified + + @staticmethod + def verify_signature( + signer_info, matching_certificate, signed_data, crypto_hash_algorithm + ): + matching_certificate_verified = None + signature = signer_info['signature'].native + + # Load the certificate using asn1crypto as it can handle more cases (v1-only-with-rsa-1024-cert-not-der.apk) + cert = x509.Certificate.load(matching_certificate.chosen.dump()) + public_key_info = cert.public_key + + # Convert the ASN.1 public key to a cryptography-compatible object + public_key_der = public_key_info.dump() + public_key = serialization.load_der_public_key( + public_key_der, backend=default_backend() + ) + + try: + # RSA Key + if isinstance(public_key, rsa.RSAPublicKey): + public_key.verify( + signature, + signed_data, + padding.PKCS1v15(), + crypto_hash_algorithm(), + ) + + # DSA Key + elif isinstance(public_key, dsa.DSAPublicKey): + public_key.verify( + signature, signed_data, crypto_hash_algorithm() + ) + + # EC Key + elif isinstance(public_key, ec.EllipticCurvePublicKey): + public_key.verify( + signature, signed_data, ec.ECDSA(crypto_hash_algorithm()) + ) + + else: + raise ValueError( + f"Unsupported key algorithm: {public_key.__class__.__name__.lower()}" + ) + + # If verification succeeds, return the certificate + matching_certificate_verified = matching_certificate.chosen.dump() + + except InvalidSignature: + logger.info( + f"The public key of the certificate: {hashlib.sha256(matching_certificate.chosen.dump()).hexdigest()} " + f"is not associated with the signature!" + ) + + return matching_certificate_verified + + @staticmethod + def get_hash_algorithm(signer_info): + # Determine the hash algorithm from the SignerInfo + digest_algorithm = signer_info['digest_algorithm']['algorithm'].native + # Map the digest algorithm to a hash function + hash_algorithms = { + 'md5': (md5, hashes.MD5), + 'sha1': (sha1, hashes.SHA1), + 'sha224': (sha224, hashes.SHA224), + 'sha256': (sha256, hashes.SHA256), + 'sha384': (sha384, hashes.SHA384), + 'sha512': (sha512, hashes.SHA512), + } + if digest_algorithm not in hash_algorithms: + raise ValueError(f"Unsupported hash algorithm: {digest_algorithm}") + return hash_algorithms[digest_algorithm] + + def find_certificate(self, signed_data_certificates, signer_info): + """ + From the bag of certs, obtain the certificate referenced by the SignerInfo. + + Args: + signed_data_certificates: List of certificates in the SignedData. + signer_info: SignerInfo object containing the issuer and serial number reference. + + Returns: + The matching certificate if found, otherwise None. + """ + matching_certificate = None + issuer_and_serial_number = signer_info['sid'] + issuer_str = self.canonical_name( + issuer_and_serial_number.chosen['issuer'] + ) + serial_number = issuer_and_serial_number.native['serial_number'] + + # # Create a x509.Name object for the issuer in the SignerInfo + # issuer_name = x509.Name.build(issuer) + # issuer_str = self.canonical_name(issuer_name) + + for cert in signed_data_certificates: + if cert.name == 'certificate': + cert_issuer = self.canonical_name( + cert.chosen['tbs_certificate']['issuer'] + ) + cert_serial_number = cert.native['tbs_certificate'][ + 'serial_number' + ] + + # Compare the canonical string representations of the issuers and the serial numbers + if ( + cert_issuer == issuer_str + and cert_serial_number == serial_number + ): + matching_certificate = cert + break + + return matching_certificate + + def get_certificate(self, filename: str) -> Union[x509.Certificate, None]: """ Return a X.509 certificate object by giving the name in the apk file @@ -1544,21 +1996,151 @@ class APK: :returns: a :class:`Certificate` certificate """ cert = self.get_certificate_der(filename) - certificate = x509.Certificate.load(cert) - + if cert: + certificate = x509.Certificate.load(cert) + else: + certificate = None return certificate - def new_zip(self, filename, deleted_files=None, new_files={}): + def canonical_name(self, name: Any, android: bool = False) -> str: """ - Create a new zip file + /* + * Method is dual-licensed under the Apache License 2.0 and GPLv3+. + * The original author has granted permission to use this code snippet under the + * Apache License 2.0 for inclusion in this project. + * https://github.com/obfusk/x509_canonical_name.py/blob/master/x509_canonical_name.py + */ + Canonical representation of x509.Name as str (with raw control characters + in places those are not stripped by normalisation). + """ + # return ",".join("+".join(f"{t}:{v}" for _, t, v in avas) for avas in self.comparison_name(name)) + return ",".join( + "+".join(f"{t}={v}" for t, v in avas) + for avas in self.comparison_name(name, android=android) + ) - :param filename: the output filename of the zip - :param deleted_files: a regex pattern to remove specific file - :param new_files: a dictionnary of new files + def comparison_name( + self, name: x509.Name, *, android: bool = False + ) -> List[List[Tuple[str, str]]]: + """ + /* + * Method is dual-licensed under the Apache License 2.0 and GPLv3+. + * The original author has granted permission to use this code snippet under the + * Apache License 2.0 for inclusion in this project. + * https://github.com/obfusk/x509_canonical_name.py/blob/master/x509_canonical_name.py + */ + Canonical representation of x509.Name as nested list. - :type filename: string - :type deleted_files: None or a string - :type new_files: a dictionnary (key:filename, value:content of the file) + Returns a list of RDNs which are a list of AVAs which are a (type, value) + tuple, where type is the standard name or dotted OID, and value is the + normalised string representation of the value. + """ + + return [ + [(t, nv) for _, t, nv, _ in avas] + for avas in self.x509_ordered_name(name, android=android) + ] + + @staticmethod + def x509_ordered_name( + name: x509.Name, + *, # type: ignore[no-any-unimported] + android: bool = False, + ) -> List[List[Tuple[int, str, str, str]]]: + """ + /* + * Method is dual-licensed under the Apache License 2.0 and GPLv3+. + * The original author has granted permission to use this code snippet under the + * Apache License 2.0 for inclusion in this project. + * https://github.com/obfusk/x509_canonical_name.py/blob/master/x509_canonical_name.py + */ + Representation of x509.Name as nested list, in canonical ordering (but also + including non-canonical pre-normalised string values). + + Returns a list of RDNs which are a list of AVAs which are a (oid, type, + normalised_value, esc_value) tuple, where oid is 0 for standard names and 1 + for dotted OIDs, type is the standard name or dotted OID, normalised_value + is the normalised string representation of the value, and esc_value is the + string value before normalisation (but after escaping). + + NB: control characters are not escaped, only characters in ",+<>;\"\\" and + "#" at the start (before "whitespace" trimming) are. + + https://docs.oracle.com/en/java/javase/21/docs/api/java.base/javax/security/auth/x500/X500Principal.html#getName(java.lang.String) + https://github.com/openjdk/jdk/blob/jdk-21%2B35/src/java.base/share/classes/sun/security/x509/AVA.java#L805 + https://github.com/openjdk/jdk/blob/jdk-21%2B35/src/java.base/share/classes/sun/security/x509/RDN.java#L472 + https://android.googlesource.com/platform/libcore/+/refs/heads/android14-release/ojluni/src/main/java/sun/security/x509/RDN.java#481 + """ + + def key( + ava: Tuple[int, str, str, str] + ) -> Tuple[int, Union[str, List[int]], str]: + o, t, nv, _ = ava + if android and o: + return o, [int(x) for x in t.split(".")], nv + return o, t, nv + + DS, U8, PS = ( + x509.DirectoryString, + x509.UTF8String, + x509.PrintableString, + ) + oids = { + "2.5.4.3": ("common_name", "cn"), + "2.5.4.6": ("country_name", "c"), + "2.5.4.7": ("locality_name", "l"), + "2.5.4.8": ("state_or_province_name", "st"), + "2.5.4.9": ("street_address", "street"), + "2.5.4.10": ("organization_name", "o"), + "2.5.4.11": ("organizational_unit_name", "ou"), + "0.9.2342.19200300.100.1.1": ("user_id", "uid"), + "0.9.2342.19200300.100.1.25": ("domain_component", "dc"), + } + esc = {ord(c): f"\\{c}" for c in ",+<>;\"\\"} + cws = "".join( + chr(i) for i in range(32 + 1) + ) # control (but not esc) and whitespace + data = [] + for rdn in reversed(name.chosen): + avas = [] + for ava in rdn: + at, av = ava["type"], ava["value"] + if at.dotted in oids: + o, t = 0, oids[at.dotted][1] # order standard before OID + else: + o, t = 1, at.dotted + if o or not ( + isinstance(av, DS) and isinstance(av.chosen, (U8, PS)) + ): + ev = nv = "#" + binascii.hexlify(av.dump()).decode() + else: + ev = (av.native or "").translate(esc) + if ev.startswith("#"): + ev = "\\" + ev + nv = unicodedata.normalize( + "NFKD", + re.sub(r" +", " ", ev).strip(cws).upper().lower(), + ) + avas.append((o, t, nv, ev)) + data.append(sorted(avas, key=key)) + return data + + def new_zip( + self, + filename: str, + deleted_files: Union[str, None] = None, + new_files: dict = {}, + ) -> None: + """ + Create a new zip file + + :param filename: the output filename of the zip + :param deleted_files: a regex pattern to remove specific file + :param new_files: a dictionnary of new files + + :type filename: string + :type deleted_files: None or a string + :type new_files: a dictionnary (key:filename, value:content of the file) """ zout = zipfile.ZipFile(filename, 'w') @@ -1589,18 +2171,18 @@ class APK: zout.writestr(item, buffer) zout.close() - def get_android_manifest_axml(self): + def get_android_manifest_axml(self) -> Union[AXMLPrinter, None]: """ - Return the :class:`AXMLPrinter` object which corresponds to the AndroidManifest.xml file + Return the :class:`AXMLPrinter` object which corresponds to the AndroidManifest.xml file - :rtype: :class:`~androguard.core.bytecodes.axml.AXMLPrinter` + :rtype: :class:`~androguard.core.axml.AXMLPrinter` """ try: return self.axml["AndroidManifest.xml"] except KeyError: return None - def get_android_manifest_xml(self): + def get_android_manifest_xml(self) -> Union[lxml.etree.Element, None]: """ Return the parsed xml object which corresponds to the AndroidManifest.xml file @@ -1611,11 +2193,11 @@ class APK: except KeyError: return None - def get_android_resources(self): + def get_android_resources(self) -> Union[ARSCParser, None]: """ - Return the :class:`~androguard.core.bytecodes.axml.ARSCParser` object which corresponds to the resources.arsc file + Return the :class:`~androguard.core.axml.ARSCParser` object which corresponds to the resources.arsc file - :rtype: :class:`~androguard.core.bytecodes.axml.ARSCParser` + :rtype: :class:`~androguard.core.axml.ARSCParser` """ try: return self.arsc["resources.arsc"] @@ -1624,16 +2206,20 @@ class APK: # There is a rare case, that no resource file is supplied. # Maybe it was added manually, thus we check here return None - self.arsc["resources.arsc"] = ARSCParser(self.zip.read("resources.arsc")) + self.arsc["resources.arsc"] = ARSCParser( + self.zip.read("resources.arsc") + ) return self.arsc["resources.arsc"] - def is_signed(self): + def is_signed(self) -> bool: """ Returns true if any of v1, v2, or v3 signatures were found. """ - return self.is_signed_v1() or self.is_signed_v2() or self.is_signed_v3() + return ( + self.is_signed_v1() or self.is_signed_v2() or self.is_signed_v3() + ) - def is_signed_v1(self): + def is_signed_v1(self) -> bool: """ Returns true if a v1 / JAR signature was found. @@ -1642,7 +2228,7 @@ class APK: """ return self.get_signature_name() is not None - def is_signed_v2(self): + def is_signed_v2(self) -> bool: """ Returns true of a v2 / APK signature was found. @@ -1654,7 +2240,7 @@ class APK: return self._is_signed_v2 - def is_signed_v3(self): + def is_signed_v3(self) -> bool: """ Returns true of a v3 / APK signature was found. @@ -1666,12 +2252,14 @@ class APK: return self._is_signed_v3 - def read_uint32_le(self, io_stream): - value, = unpack(' int: + (value,) = unpack(' list[tuple[int, bytes]]: + """Parse digests""" if not len(digest_bytes): return [] @@ -1690,7 +2278,7 @@ class APK: return digests - def parse_v2_v3_signature(self): + def parse_v2_v3_signature(self) -> None: # Need to find an v2 Block in the APK. # The Google Docs gives you the following rule: # * go to the end of the ZIP File @@ -1714,11 +2302,17 @@ class APK: while f.tell() > 0: f.seek(-1, io.SEEK_CUR) - r, = unpack('<4s', f.read(4)) + (r,) = unpack('<4s', f.read(4)) if r == self._PK_END_OF_CENTRAL_DIR: # Read central dir - this_disk, disk_central, this_entries, total_entries, \ - size_central, offset_central = unpack(' None: """ Parse the V2 signing block and extract all features """ @@ -1805,7 +2408,9 @@ class APK: # * publickey size_sequence = self.read_uint32_le(block) if size_sequence + 4 != len(block_bytes): - raise BrokenAPKError("size of sequence and blocksize does not match") + raise BrokenAPKError( + "size of sequence and blocksize does not match" + ) while block.tell() < len(block_bytes): off_signer = block.tell() @@ -1862,7 +2467,7 @@ class APK: publickey = block.read(len_publickey) signer = APKV3Signer() - signer._bytes = view[off_signer:off_signer + size_signer] + signer._bytes = view[off_signer: off_signer + size_signer] signer.signed_data = signed_data_object signer.signatures = sigs signer.public_key = publickey @@ -1871,7 +2476,7 @@ class APK: self._v3_signing_data.append(signer) - def parse_v2_signing_block(self): + def parse_v2_signing_block(self) -> None: """ Parse the V2 signing block and extract all features """ @@ -1900,7 +2505,9 @@ class APK: size_sequence = self.read_uint32_le(block) if size_sequence + 4 != len(block_bytes): - raise BrokenAPKError("size of sequence and blocksize does not match") + raise BrokenAPKError( + "size of sequence and blocksize does not match" + ) while block.tell() < len(block_bytes): off_signer = block.tell() @@ -1946,14 +2553,14 @@ class APK: publickey = block.read(len_publickey) signer = APKV2Signer() - signer._bytes = view[off_signer:off_signer + size_signer] + signer._bytes = view[off_signer: off_signer + size_signer] signer.signed_data = signed_data_object signer.signatures = sigs signer.public_key = publickey self._v2_signing_data.append(signer) - def get_public_keys_der_v3(self): + def get_public_keys_der_v3(self) -> list[bytes]: """ Return a list of DER coded X.509 public keys from the v3 signature block """ @@ -1968,7 +2575,7 @@ class APK: return public_keys - def get_public_keys_der_v2(self): + def get_public_keys_der_v2(self) -> list[bytes]: """ Return a list of DER coded X.509 public keys from the v3 signature block """ @@ -1983,7 +2590,7 @@ class APK: return public_keys - def get_certificates_der_v3(self): + def get_certificates_der_v3(self) -> list[bytes]: """ Return a list of DER coded X.509 certificates from the v3 signature block """ @@ -1992,13 +2599,15 @@ class APK: self.parse_v3_signing_block() certs = [] - for signed_data in [signer.signed_data for signer in self._v3_signing_data]: + for signed_data in [ + signer.signed_data for signer in self._v3_signing_data + ]: for cert in signed_data.certificates: certs.append(cert) return certs - def get_certificates_der_v2(self): + def get_certificates_der_v2(self) -> list[bytes]: """ Return a list of DER coded X.509 certificates from the v3 signature block """ @@ -2007,75 +2616,93 @@ class APK: self.parse_v2_signing_block() certs = [] - for signed_data in [signer.signed_data for signer in self._v2_signing_data]: + for signed_data in [ + signer.signed_data for signer in self._v2_signing_data + ]: for cert in signed_data.certificates: certs.append(cert) return certs - def get_public_keys_v3(self): + def get_public_keys_v3(self) -> list[keys.PublicKeyInfo]: """ Return a list of :class:`asn1crypto.keys.PublicKeyInfo` which are found in the v3 signing block. """ - return [keys.PublicKeyInfo.load(pkey) for pkey in self.get_public_keys_der_v3()] + return [ + keys.PublicKeyInfo.load(pkey) + for pkey in self.get_public_keys_der_v3() + ] - def get_public_keys_v2(self): + def get_public_keys_v2(self) -> list[keys.PublicKeyInfo]: """ Return a list of :class:`asn1crypto.keys.PublicKeyInfo` which are found in the v2 signing block. """ - return [keys.PublicKeyInfo.load(pkey) for pkey in self.get_public_keys_der_v2()] + return [ + keys.PublicKeyInfo.load(pkey) + for pkey in self.get_public_keys_der_v2() + ] - def get_certificates_v3(self): + def get_certificates_v3(self) -> list[x509.Certificate]: """ Return a list of :class:`asn1crypto.x509.Certificate` which are found in the v3 signing block. Note that we simply extract all certificates regardless of the signer. Therefore this is just a list of all certificates found in all signers. """ - return [x509.Certificate.load(cert) for cert in self.get_certificates_der_v3()] + return [ + x509.Certificate.load(cert) + for cert in self.get_certificates_der_v3() + ] - def get_certificates_v2(self): + def get_certificates_v2(self) -> list[x509.Certificate]: """ Return a list of :class:`asn1crypto.x509.Certificate` which are found in the v2 signing block. Note that we simply extract all certificates regardless of the signer. Therefore this is just a list of all certificates found in all signers. """ - return [x509.Certificate.load(cert) for cert in self.get_certificates_der_v2()] + return [ + x509.Certificate.load(cert) + for cert in self.get_certificates_der_v2() + ] - def get_certificates_v1(self): + def get_certificates_v1(self) -> list[Union[x509.Certificate, None]]: """ - Return a list of :class:`asn1crypto.x509.Certificate` which are found + Return a list of verified :class:`asn1crypto.x509.Certificate` which are found in the META-INF folder (v1 signing). - Note that we simply extract all certificates regardless of the signer. - Therefore this is just a list of all certificates found in all signers. """ certs = [] for x in self.get_signature_names(): - certs.append(x509.Certificate.load(self.get_certificate_der(x))) - + cc = self.get_certificate_der(x) + if cc is not None: + certs.append(x509.Certificate.load(cc)) return certs - def get_certificates(self): + def get_certificates(self) -> list[x509.Certificate]: """ Return a list of unique :class:`asn1crypto.x509.Certificate` which are found in v1, v2 and v3 signing Note that we simply extract all certificates regardless of the signer. Therefore this is just a list of all certificates found in all signers. + Exception is v1, for which the certificate returned is verified. """ fps = [] certs = [] - for x in self.get_certificates_v1() + self.get_certificates_v2() + self.get_certificates_v3(): + for x in ( + self.get_certificates_v1() + + self.get_certificates_v2() + + self.get_certificates_v3() + ): if x.sha256 not in fps: fps.append(x.sha256) certs.append(x) return certs - def get_signature_name(self): + def get_signature_name(self) -> Union[str, None]: """ - Return the name of the first signature file found. + Return the name of the first signature file found. """ if self.get_signature_names(): return self.get_signature_names()[0] @@ -2083,7 +2710,7 @@ class APK: # Unsigned APK return None - def get_signature_names(self): + def get_signature_names(self) -> list[str]: """ Return a list of the signature file names (v1 Signature / JAR Signature) @@ -2099,11 +2726,14 @@ class APK: signatures.append(i) else: logger.warning( - "v1 signature file {} missing .SF file - Partial signature!".format(i)) + "v1 signature file {} missing .SF file - Partial signature!".format( + i + ) + ) return signatures - def get_signature(self): + def get_signature(self) -> Union[str, None]: """ Return the data of the first signature file found (v1 Signature / JAR Signature) @@ -2115,7 +2745,7 @@ class APK: else: return None - def get_signatures(self): + def get_signatures(self) -> list[bytes]: """ Return a list of the data of the signature files. Only v1 / JAR Signing. @@ -2131,7 +2761,7 @@ class APK: return signature_datas - def show(self): + def show(self) -> None: self.get_files_types() print("FILES: ") @@ -2184,23 +2814,31 @@ class APK: show_Certificate(c) -def show_Certificate(cert, short=False): +def show_Certificate(cert, short: bool = False) -> None: """ - Print Fingerprints, Issuer and Subject of an X509 Certificate. + Print Fingerprints, Issuer and Subject of an X509 Certificate. - :param cert: X509 Certificate to print - :param short: Print in shortform for DN (Default: False) + :param cert: X509 Certificate to print + :param short: Print in shortform for DN (Default: False) - :type cert: :class:`asn1crypto.x509.Certificate` - :type short: Boolean + :type cert: :class:`asn1crypto.x509.Certificate` + :type short: Boolean """ print("SHA1 Fingerprint: {}".format(cert.sha1_fingerprint)) print("SHA256 Fingerprint: {}".format(cert.sha256_fingerprint)) - print("Issuer: {}".format(get_certificate_name_string(cert.issuer.native, short=short))) - print("Subject: {}".format(get_certificate_name_string(cert.subject.native, short=short))) + print( + "Issuer: {}".format( + get_certificate_name_string(cert.issuer.native, short=short) + ) + ) + print( + "Subject: {}".format( + get_certificate_name_string(cert.subject.native, short=short) + ) + ) -def ensure_final_value(packageName, arsc, value): +def ensure_final_value(packageName: str, arsc: ARSCParser, value: str) -> str: """Ensure incoming value is always the value, not the resid androguard will sometimes return the Android "resId" aka @@ -2221,3 +2859,66 @@ def ensure_final_value(packageName, arsc, value): pass return returnValue return '' + + +def get_apkid(apkfile: str) -> tuple[str, str, str]: + """Read (appid, versionCode, versionName) from an APK + + This first tries to do quick binary XML parsing to just get the + values that are needed. It will fallback to full androguard + parsing, which is slow, if it can't find the versionName value or + versionName is set to a Android String Resource (e.g. an integer + hex value that starts with @). + + """ + logger.debug("GET_APKID") + + if not os.path.exists(apkfile): + logger.error("'{apkfile}' does not exist!".format(apkfile=apkfile)) + + appid = None + versionCode = None + versionName = None + apk = ZipEntry.parse(apkfile, False) + manifest = apk.read('AndroidManifest.xml') + axml = AXMLParser(manifest) + count = 0 + while axml.is_valid(): + _type = next(axml) + count += 1 + if _type == START_TAG: + for i in range(0, axml.getAttributeCount()): + name = axml.getAttributeName(i) + _type = axml.getAttributeValueType(i) + _data = axml.getAttributeValueData(i) + value = format_value( + _type, _data, lambda _: axml.getAttributeValue(i) + ) + if appid is None and name == 'package': + appid = value + elif versionCode is None and name == 'versionCode': + if value.startswith('0x'): + versionCode = str(int(value, 16)) + else: + versionCode = value + elif versionName is None and name == 'versionName': + versionName = value + + if axml.name == 'manifest': + break + elif _type == END_TAG or _type == TEXT or _type == END_DOCUMENT: + raise RuntimeError( + '{path}: must be the first element in AndroidManifest.xml'.format( + path=apkfile + ) + ) + + if not versionName or versionName[0] == '@': + a = APK(apkfile) + versionName = ensure_final_value( + a.package, a.get_android_resources(), a.get_androidversion_name() + ) + if not versionName: + versionName = '' # versionName is expected to always be a str + + return appid, versionCode, versionName.strip('\0') diff --git a/mobsf/StaticAnalyzer/tools/androguard4/apkinspector/extract.py b/mobsf/StaticAnalyzer/tools/androguard4/apkinspector/extract.py new file mode 100644 index 00000000..697b09b4 --- /dev/null +++ b/mobsf/StaticAnalyzer/tools/androguard4/apkinspector/extract.py @@ -0,0 +1,106 @@ +# -*- coding: utf_8 -*- +# flake8: noqa +import zlib +import os + + +def extract_file_based_on_header_info(apk_file, local_header_info, central_directory_info): + """ + Extracts a single file from the apk_file based on the information provided from the offset and the header_info. + It takes into account that the compression method provided might not be STORED or DEFLATED! The returned + 'indicator', shows what compression method was used. Besides the standard STORED/DEFLATE it may return + 'DEFLATED_TAMPERED', which means that the compression method found was not DEFLATED(8) but it should have been, + and 'STORED_TAMPERED' which means that the compression method found was not STORED(0) but should have been. + + :param apk_file: The APK file e.g. with open('test.apk', 'rb') as apk_file + :type apk_file: bytesIO + :param local_header_info: The local header dictionary info for that specific filename + :type local_header_info: dict + :param central_directory_info: The central directory entry for that specific filename + :type central_directory_info: dict + :return: Returns the actual extracted data for that file along with an indication of whether a static analysis evasion technique was used or not. + :rtype: set(bytes, str) + """ + filename_length = local_header_info["file_name_length"] + if local_header_info["compressed_size"] == 0 or local_header_info["uncompressed_size"] == 0: + compressed_size = central_directory_info["compressed_size"] + uncompressed_size = central_directory_info["uncompressed_size"] + else: + compressed_size = local_header_info["compressed_size"] + uncompressed_size = local_header_info["uncompressed_size"] + + extra_field_length = local_header_info["extra_field_length"] + compression_method = local_header_info["compression_method"] + # Skip the offset + local header to reach the compressed data + local_header_size = 30 # Size of the local header in bytes + offset = central_directory_info["relative_offset_of_local_file_header"] + apk_file.seek(offset + local_header_size + filename_length + extra_field_length) + if compression_method == 0: # Stored (no compression) + uncompressed_data = apk_file.read(uncompressed_size) + extracted_data = uncompressed_data + indicator = 'STORED' + elif compression_method == 8: + compressed_data = apk_file.read(compressed_size) + # -15 for windows size due to raw stream with no header or trailer + extracted_data = zlib.decompress(compressed_data, -15) + indicator = 'DEFLATED' + elif compressed_size == uncompressed_size: + compressed_data = apk_file.read(uncompressed_size) + extracted_data = compressed_data + indicator = 'STORED_TAMPERED' + else: + cur_loc = apk_file.tell() + try: + compressed_data = apk_file.read(compressed_size) + extracted_data = zlib.decompress(compressed_data, -15) + indicator = 'DEFLATED_TAMPERED' + except: + apk_file.seek(cur_loc) + compressed_data = apk_file.read(uncompressed_size) + extracted_data = compressed_data + indicator = 'STORED_TAMPERED' + return extracted_data, indicator + + +def extract_all_files_from_central_directory(apk_file, central_directory_entries, local_header_entries, output_dir): + """ + Extracts all files from an APK based on the entries detected in the central_directory_entries. + + :param apk_file: The APK file e.g. with open('test.apk', 'rb') as apk_file + :type apk_file: bytesIO + :param central_directory_entries: The dictionary with all the entries for the central directory + :type central_directory_entries: dict + :param local_header_entries: The dictionary with all the local header entries + :type local_header_entries: dict + :param output_dir: The output directory where to save the files. + :type output_dir: str + :return: Returns 0 if no errors, 1 if an exception and 2 if the output directory already exists + :rtype: int + """ + try: + # Check if the output directory already exists + if os.path.exists(output_dir): + print("Extraction aborted. Output directory already exists.") + return 2 + # Create the output directory or overwrite if it already exists + os.makedirs(output_dir, exist_ok=True) + # Iterate over central directory entries + for filename, cd_header_info in central_directory_entries.items(): + if not filename: + # to account for the cases where an empty filename entry is added + continue + # Extract the file using the local header information + extracted_data = \ + extract_file_based_on_header_info( + apk_file, local_header_entries[filename], cd_header_info)[0] + # Construct the output file path + output_path = os.path.join(output_dir, filename) + # Create directories if necessary + os.makedirs(os.path.dirname(output_path), exist_ok=True) + # Write the extracted data to the output file + with open(output_path, 'wb') as output_file: + output_file.write(extracted_data) + return 0 + except Exception as e: + print(f"Error extracting files: {e}") + return 1 diff --git a/mobsf/StaticAnalyzer/tools/androguard4/zipfile.py b/mobsf/StaticAnalyzer/tools/androguard4/apkinspector/headers.py similarity index 85% rename from mobsf/StaticAnalyzer/tools/androguard4/zipfile.py rename to mobsf/StaticAnalyzer/tools/androguard4/apkinspector/headers.py index eef87190..a9d3096c 100644 --- a/mobsf/StaticAnalyzer/tools/androguard4/zipfile.py +++ b/mobsf/StaticAnalyzer/tools/androguard4/apkinspector/headers.py @@ -1,64 +1,13 @@ # -*- coding: utf_8 -*- # flake8: noqa -"""This file is from apkinspector licensed under the Apache License 2.0.""" +# ApkInspector - Nov 24, 2024 - 293ab2d89ab9ce011c7dbbc5df3c876172875a1c import io -import zlib +import os import struct from typing import Dict - -def extract_file_based_on_header_info(apk_file, local_header_info, central_directory_info): - """ - Extracts a single file from the apk_file based on the information provided from the offset and the header_info. - It takes into account that the compression method provided might not be STORED or DEFLATED! The returned - 'indicator', shows what compression method was used. Besides the standard STORED/DEFLATE it may return - 'DEFLATED_TAMPERED', which means that the compression method found was not DEFLATED(8) but it should have been, - and 'STORED_TAMPERED' which means that the compression method found was not STORED(0) but should have been. - - :param apk_file: The APK file e.g. with open('test.apk', 'rb') as apk_file - :type apk_file: bytesIO - :param local_header_info: The local header dictionary info for that specific filename - :type local_header_info: dict - :param central_directory_info: The central directory entry for that specific filename - :type central_directory_info: dict - :return: Returns the actual extracted data for that file along with an indication of whether a static analysis evasion technique was used or not. - :rtype: set(bytes, str) - """ - filename_length = local_header_info["file_name_length"] - if local_header_info["compressed_size"] == 0 or local_header_info["uncompressed_size"] == 0: - compressed_size = central_directory_info["compressed_size"] - uncompressed_size = central_directory_info["uncompressed_size"] - else: - compressed_size = local_header_info["compressed_size"] - uncompressed_size = local_header_info["uncompressed_size"] - - extra_field_length = local_header_info["extra_field_length"] - compression_method = local_header_info["compression_method"] - # Skip the offset + local header to reach the compressed data - local_header_size = 30 # Size of the local header in bytes - offset = central_directory_info["relative_offset_of_local_file_header"] - apk_file.seek(offset + local_header_size + filename_length + extra_field_length) - if compression_method == 0: # Stored (no compression) - uncompressed_data = apk_file.read(uncompressed_size) - extracted_data = uncompressed_data - indicator = 'STORED' - elif compression_method == 8: - compressed_data = apk_file.read(compressed_size) - # -15 for windows size due to raw stream with no header or trailer - extracted_data = zlib.decompress(compressed_data, -15) - indicator = 'DEFLATED' - else: - try: - cur_loc = apk_file.tell() - compressed_data = apk_file.read(compressed_size) - extracted_data = zlib.decompress(compressed_data, -15) - indicator = 'DEFLATED_TAMPERED' - except: - apk_file.seek(cur_loc) - compressed_data = apk_file.read(uncompressed_size) - extracted_data = compressed_data - indicator = 'STORED_TAMPERED' - return extracted_data, indicator +from .extract import extract_file_based_on_header_info, extract_all_files_from_central_directory +from .helpers import pretty_print_header, save_to_json, save_data_to_file class EndOfCentralDirectoryRecord: @@ -98,19 +47,18 @@ class EndOfCentralDirectoryRecord: signature_offset = -1 file_size = apk_file.seek(0, 2) while offset < file_size: - position = file_size - offset - chunk_size - if position < 0: - position = 0 + position = max(0, file_size - offset - chunk_size) apk_file.seek(position) chunk = apk_file.read(chunk_size) if not chunk: break - # end of Central Directory File Header signature - signature_offset = chunk.rfind(b'\x50\x4b\x05\x06') + signature_offset = chunk.rfind(b'\x50\x4b\x05\x06') # EOCD signature if signature_offset != -1: eo_central_directory_offset = position + signature_offset - break # Found End of central directory record (EOCD) signature - offset += chunk_size + break # Found EOCD signature + # Adjust offset to overlap by 4 bytes + offset += chunk_size - 4 + if signature_offset == -1: raise ValueError( "End of central directory record (EOCD) signature not found") @@ -301,11 +249,11 @@ class CentralDirectory: relative_offset_of_local_file_header = struct.unpack(' Dict[str, CentralDirectoryEntry]: @@ -582,3 +534,63 @@ class ZipEntry: :rtype: list """ return [vl for vl in self.central_directory.to_dict()] + + def extract_all(self, extract_path, apk_name): + """ + Extracts all the contents of the APK. + + :param extract_path: where to extract it + :type extract_path: str + :param apk_name: the name of the apk + :type apk_name: str + """ + output_path = os.path.join(extract_path, apk_name) + if not extract_all_files_from_central_directory(self.zip, self.to_dict()["central_directory"], + self.to_dict()["local_headers"], output_path): + print(f"Extraction successful for: {apk_name}") + + +def print_headers_of_filename(cd_h_of_file, local_header_of_file): + """ + Prints out the details for both the central directory header and the local file header. Useful for the CLI. + + :param cd_h_of_file: central directory header of a filename as it may be retrieved from headers_of_filename + :type cd_h_of_file: dict + :param local_header_of_file: local header dictionary of a filename as it may be retrieved from headers_of_filename + :type local_header_of_file: dict + """ + if not cd_h_of_file or not local_header_of_file: + print("Are you sure the filename exists?") + return + pretty_print_header("CENTRAL DIRECTORY") + for k in cd_h_of_file: + if k == 'Relative offset of local file header' or k == 'Offset in the central directory header': + print(f"{k:40} : {hex(int(cd_h_of_file[k]))} | {cd_h_of_file[k]}") + else: + print(f"{k:40} : {cd_h_of_file[k]}") + pretty_print_header("LOCAL HEADER") + for k in local_header_of_file: + print(f"{k:40} : {local_header_of_file[k]}") + + +def show_and_save_info_of_headers(entries, apk_name, header_type: str, export: bool, show: bool): + """ + Print information for each entry for the central directory header and allow to possibly export to JSON. + + :param entries: The dictionary with all the entries for the central directory + :type entries: dict + :param apk_name: String with the name of the APK, so it can be used for the export. + :type apk_name: str + :param header_type: What type of header that is, either central_directory or local, to be used for the export + :type header_type: str + :param export: Boolean for exporting or not to JSON + :type export: bool + :param show: Boolean for printing or not the entries + :type show: bool + """ + if show: + for entry in entries: + pretty_print_header(entry) + print(entries[entry]) + if export: + save_to_json(f"{apk_name}_{header_type}_header.json", entries) diff --git a/mobsf/StaticAnalyzer/tools/androguard4/apkinspector/helpers.py b/mobsf/StaticAnalyzer/tools/androguard4/apkinspector/helpers.py new file mode 100644 index 00000000..44213a0e --- /dev/null +++ b/mobsf/StaticAnalyzer/tools/androguard4/apkinspector/helpers.py @@ -0,0 +1,68 @@ +# -*- coding: utf_8 -*- +# flake8: noqa +import json + + +def pretty_print_header(header_text, width=50, char='-'): + """ + Formatting output used for the CLI + + :param header_text: The text to be displayed + :type header_text: str + :param width: total width of the display + :type width: int + :param char: which char to be used as a filler + :type char: str + """ + padding = max(0, width - len(header_text)) // 2 + formatted_header = f"\n{char * padding} {header_text} {char * padding}" + print(formatted_header) + + +def save_data_to_file(filename, data): + """ + Write data to file + + :param data: the actual data + :type data: bytes + :param filename: file to be saved in + :type filename: str + """ + try: + with open(filename, 'wb') as output_file: + output_file.write(data) + print(f"Data saved to {filename}") + except Exception as e: + print(f"Error while saving data to {filename}: {e}") + + +def save_to_json(filename, dictionary): + """ + Simple method to save a dictionary as JSON into the filename. + + :param filename: the name of the file to be saved as + :type filename: str + :param dictionary: the dictionary to be saved as JSON + :type dictionary: dict + """ + with open(filename, "w") as h_file: + json.dump(dictionary, h_file, indent=4) + + +def escape_xml_entities(data): + """ + Escaping characters that cant be included within an XML file. + + :param data: The string to escape + :type data: str + :return: The escaped output + :rtype: str + """ + replacements = { + '<': '<', + '>': '>', + '&': '&', + '"': '"', + "'": ''' + } + return ''.join(replacements.get(c, c) for c in data) diff --git a/mobsf/StaticAnalyzer/tools/androguard4/axml.py b/mobsf/StaticAnalyzer/tools/androguard4/axml.py index b7dbf529..c85c2de5 100644 --- a/mobsf/StaticAnalyzer/tools/androguard4/axml.py +++ b/mobsf/StaticAnalyzer/tools/androguard4/axml.py @@ -1,22 +1,32 @@ # -*- coding: utf_8 -*- # flake8: noqa +# Androguard4 AXML - Nov 24, 2024 -04a5703b8ba7c181bb9f5f5995a2c16b6f9353cf +# Allows type hinting of types not-yet-declared +# in Python >= 3.7 +# see https://peps.python.org/pep-0563/ +from __future__ import annotations + +import binascii +import collections +import io +import re +from collections import defaultdict +from struct import pack, unpack +from typing import BinaryIO, Union +from xml.sax.saxutils import escape + +from lxml import etree + +import logging + from .resources import public from .types import * -from struct import pack, unpack -from xml.sax.saxutils import escape -import collections -from collections import defaultdict - -from lxml import etree -import re -import binascii -import io -import logging logger = logging.getLogger(__name__) logger.setLevel(level=logging.CRITICAL) + # Constants for ARSC Files # see http://aospxref.com/android-13.0.0_r3/xref/frameworks/base/libs/androidfw/include/androidfw/ResourceTypes.h#233 RES_NULL_TYPE = 0x0000 @@ -30,7 +40,7 @@ RES_XML_END_NAMESPACE_TYPE = 0x0101 RES_XML_START_ELEMENT_TYPE = 0x0102 RES_XML_END_ELEMENT_TYPE = 0x0103 RES_XML_CDATA_TYPE = 0x0104 -RES_XML_LAST_CHUNK_TYPE = 0x017f +RES_XML_LAST_CHUNK_TYPE = 0x017F RES_XML_RESOURCE_MAP_TYPE = 0x0180 @@ -78,7 +88,7 @@ TYPE_TABLE = { TYPE_STRING: "string", } -RADIX_MULTS = [0.00390625, 3.051758E-005, 1.192093E-007, 4.656613E-010] +RADIX_MULTS = [0.00390625, 3.051758e-005, 1.192093e-007, 4.656613e-010] DIMENSION_UNITS = ["px", "dip", "sp", "pt", "in", "mm"] FRACTION_UNITS = ["%", "%p"] @@ -87,10 +97,11 @@ COMPLEX_UNIT_MASK = 0x0F class ResParserError(Exception): """Exception for the parsers""" + pass -def complexToFloat(xcomplex): +def complexToFloat(xcomplex) -> float: """ Convert a complex unit into float """ @@ -100,12 +111,12 @@ def complexToFloat(xcomplex): class StringBlock: """ StringBlock is a CHUNK inside an AXML File: `ResStringPool_header` - It contains all strings, which are used by referecing to ID's + It contains all strings, which are used by referencing to ID's See http://androidxref.com/9.0.0_r3/xref/frameworks/base/libs/androidfw/include/androidfw/ResourceTypes.h#436 """ - def __init__(self, buff, header): + def __init__(self, buff: BinaryIO, header: ARSCHeader) -> None: """ :param buff: buffer which holds the string block :param header: a instance of :class:`~ARSCHeader` @@ -120,15 +131,18 @@ class StringBlock: # flags self.flags = unpack(' 0: - logger.info("Styles Offset given, but styleCount is zero. " - "This is not a problem but could indicate packers.") + logger.info( + "Styles Offset given, but styleCount is zero. " + "This is not a problem but could indicate packers." + ) self.m_stringOffsets = [] self.m_styleOffsets = [] @@ -177,7 +193,9 @@ class StringBlock: self.m_styles.append(unpack('".format(self.stringCount, self.styleCount, self.m_isUTF8) + return "".format( + self.stringCount, self.styleCount, self.m_isUTF8 + ) def __getitem__(self, idx): """ @@ -198,7 +216,7 @@ class StringBlock: for i in range(self.stringCount): yield self.getString(i) - def getString(self, idx): + def getString(self, idx: int) -> str: """ Return the string at the index in the string table @@ -208,7 +226,7 @@ class StringBlock: if idx in self._cache: return self._cache[idx] - if idx < 0 or not self.m_stringOffsets or idx > self.stringCount: + if idx < 0 or not self.m_stringOffsets or idx >= self.stringCount: return "" offset = self.m_stringOffsets[idx] @@ -220,7 +238,7 @@ class StringBlock: return self._cache[idx] - def getStyle(self, idx): + def getStyle(self, idx: int) -> int: """ Return the style associated with the index @@ -229,7 +247,7 @@ class StringBlock: """ return self.m_styles[idx] - def _decode8(self, offset): + def _decode8(self, offset: int) -> str: """ Decode an UTF-8 String at the given offset @@ -245,15 +263,27 @@ class StringBlock: encoded_bytes, skip = self._decode_length(offset, 1) offset += skip + # Two checks should happen here: + # a) offset + encoded_bytes surpassing the string_pool length and + # b) non-null terminated strings which should be rejected + # platform/frameworks/base/libs/androidfw/ResourceTypes.cpp#789 + if len(self.m_charbuff) < (offset + encoded_bytes): + logger.warning( + f"String size: {offset + encoded_bytes} is exceeding string pool size. Returning empty string." + ) + return "" data = self.m_charbuff[offset: offset + encoded_bytes] if self.m_charbuff[offset + encoded_bytes] != 0: raise ResParserError( - "UTF-8 String is not null terminated! At offset={}".format(offset)) + "UTF-8 String is not null terminated! At offset={}".format( + offset + ) + ) return self._decode_bytes(data, 'utf-8', str_len) - def _decode16(self, offset): + def _decode16(self, offset: int) -> str: """ Decode an UTF-16 String at the given offset @@ -266,16 +296,34 @@ class StringBlock: # The len is the string len in utf-16 units encoded_bytes = str_len * 2 + # Two checks should happen here: + # a) offset + encoded_bytes surpassing the string_pool length and + # b) non-null terminated strings which should be rejected + # platform/frameworks/base/libs/androidfw/ResourceTypes.cpp#789 + if len(self.m_charbuff) < (offset + encoded_bytes): + logger.warning( + f"String size: {offset + encoded_bytes} is exceeding string pool size. Returning empty string." + ) + return "" + data = self.m_charbuff[offset: offset + encoded_bytes] - if self.m_charbuff[offset + encoded_bytes:offset + encoded_bytes + 2] != b"\x00\x00": + if ( + self.m_charbuff[ + offset + encoded_bytes: offset + encoded_bytes + 2 + ] + != b"\x00\x00" + ): raise ResParserError( - "UTF-16 String is not null terminated! At offset={}".format(offset)) + "UTF-16 String is not null terminated! At offset={}".format( + offset + ) + ) return self._decode_bytes(data, 'utf-16', str_len) @staticmethod - def _decode_bytes(data, encoding, str_len): + def _decode_bytes(data: bytes, encoding: str, str_len: int) -> str: """ Generic decoding with length check. The string is decoded from bytes with the given encoding, then the length @@ -292,7 +340,7 @@ class StringBlock: logger.warning("invalid decoded string length") return string - def _decode_length(self, offset, sizeof_char): + def _decode_length(self, offset: int, sizeof_char: int) -> tuple[int, int]: """ Generic Length Decoding at offset of string @@ -312,7 +360,9 @@ class StringBlock: fmt = "<2{}".format('B' if sizeof_char == 1 else 'H') highbit = 0x80 << (8 * (sizeof_char - 1)) - length1, length2 = unpack(fmt, self.m_charbuff[offset:(offset + sizeof_2chars)]) + length1, length2 = unpack( + fmt, self.m_charbuff[offset: (offset + sizeof_2chars)] + ) if (length1 & highbit) != 0: length = ((length1 & ~highbit) << (8 * sizeof_char)) | length2 @@ -323,28 +373,39 @@ class StringBlock: # These are true asserts, as the size should never be less than the values if sizeof_char == 1: - assert length <= 0x7FFF, "length of UTF-8 string is too large! At offset={}".format( - offset) + assert ( + length <= 0x7FFF + ), "length of UTF-8 string is too large! At offset={}".format( + offset + ) else: - assert length <= 0x7FFFFFFF, "length of UTF-16 string is too large! At offset={}".format( - offset) + assert ( + length <= 0x7FFFFFFF + ), "length of UTF-16 string is too large! At offset={}".format( + offset + ) return length, size - def show(self): + def show(self) -> None: """ Print some information on stdout about the string table """ - print("StringBlock(stringsCount=0x%x, " - "stringsOffset=0x%x, " - "stylesCount=0x%x, " - "stylesOffset=0x%x, " - "flags=0x%x" - ")" % (self.stringCount, - self.stringsOffset, - self.styleCount, - self.stylesOffset, - self.flags)) + print( + "StringBlock(stringsCount=0x%x, " + "stringsOffset=0x%x, " + "stylesCount=0x%x, " + "stylesOffset=0x%x, " + "flags=0x%x" + ")" + % ( + self.stringCount, + self.stringsOffset, + self.styleCount, + self.stylesOffset, + self.flags, + ) + ) if self.stringCount > 0: print() @@ -383,7 +444,7 @@ class AXMLParser: See http://androidxref.com/9.0.0_r3/xref/frameworks/base/libs/androidfw/include/androidfw/ResourceTypes.h#563 """ - def __init__(self, raw_buff): + def __init__(self, raw_buff: bytes) -> None: logger.debug("AXMLParser") self._reset() @@ -392,11 +453,15 @@ class AXMLParser: self.axml_tampered = False self.buff = io.BufferedReader(io.BytesIO(raw_buff)) self.buff_size = self.buff.raw.getbuffer().nbytes + self.packerwarning = False # Minimum is a single ARSCHeader, which would be a strange edge case... if self.buff_size < 8: logger.error( - "Filesize is too small to be a valid AXML file! Filesize: {}".format(self.buff_size)) + "Filesize is too small to be a valid AXML file! Filesize: {}".format( + self.buff_size + ) + ) self._valid = False return @@ -404,7 +469,10 @@ class AXMLParser: # But this is not possible as the maximum chunk size is a unsigned 4 byte int. if self.buff_size > 0xFFFFFFFF: logger.error( - "Filesize is too large to be a valid AXML file! Filesize: {}".format(self.buff_size)) + "Filesize is too large to be a valid AXML file! Filesize: {}".format( + self.buff_size + ) + ) self._valid = False return @@ -422,46 +490,66 @@ class AXMLParser: # Can be a common error: the file is not an AXML but a plain XML # The file will then usually start with ' self.buff_size: - logger.error("This does not look like an AXML file. Declared filesize does not match real size: {} vs {}".format( - self.filesize, self.buff_size)) + logger.error( + "This does not look like an AXML file. Declared filesize does not match real size: {} vs {}".format( + self.filesize, self.buff_size + ) + ) self._valid = False return if self.filesize < self.buff_size: # The file can still be parsed up to the point where the chunk should end. self.axml_tampered = True - logger.warning("Declared filesize ({}) is smaller than total file size ({}). " - "Was something appended to the file? Trying to parse it anyways.".format(self.filesize, self.buff.size())) + logger.warning( + "Declared filesize ({}) is smaller than total file size ({}). " + "Was something appended to the file? Trying to parse it anyways.".format( + self.filesize, self.buff_size + ) + ) # Not that severe of an error, we have plenty files where this is not # set correctly if axml_header.type != RES_XML_TYPE: self.axml_tampered = True - logger.warning("AXML file has an unusual resource type! " - "Malware likes to to such stuff to anti androguard! " - "But we try to parse it anyways. Resource Type: 0x{:04x}".format(axml_header.type)) + logger.warning( + "AXML file has an unusual resource type! " + "Malware likes to to such stuff to anti androguard! " + "But we try to parse it anyways. Resource Type: 0x{:04x}".format( + axml_header.type + ) + ) # Now we parse the STRING POOL try: header = ARSCHeader(self.buff, expected_type=RES_STRING_POOL_TYPE) logger.debug("STRING_POOL {}".format(header)) except ResParserError as e: - logger.error("Error parsing resource header of string pool: {}".format(e)) + logger.error( + "Error parsing resource header of string pool: {}".format(e) + ) self._valid = False return if header.header_size != 0x1C: - logger.error("This does not look like an AXML file. String chunk header size does not equal 28! header size = {}".format( - header.header_size)) + logger.error( + "This does not look like an AXML file. String chunk header size does not equal 28! header size = {}".format( + header.header_size + ) + ) self._valid = False return @@ -475,7 +563,7 @@ class AXMLParser: # Store a list of prefix/uri mappings encountered self.namespaces = [] - def is_valid(self): + def is_valid(self) -> bool: """ Get the state of the AXMLPrinter. if an error happend somewhere in the process of parsing the file, @@ -528,60 +616,89 @@ class AXMLParser: # Check size: < 8 bytes mean that the chunk is not complete # Should be aligned to 4 bytes. if h.size < 8 or (h.size % 4) != 0: - logger.error("Invalid chunk size in chunk XML_RESOURCE_MAP") + logger.error( + "Invalid chunk size in chunk XML_RESOURCE_MAP" + ) self._valid = False return for i in range((h.size - h.header_size) // 4): - self.m_resourceIDs.append(unpack(' RES_XML_LAST_CHUNK_TYPE: + if ( + h.type < RES_XML_FIRST_CHUNK_TYPE + or h.type > RES_XML_LAST_CHUNK_TYPE + ): # h.size is the size of the whole chunk including the header. # We read already 8 bytes of the header, thus we need to # subtract them. logger.error( - "Not a XML resource chunk type: 0x{:04x}. Skipping {} bytes".format(h.type, h.size)) + "Not a XML resource chunk type: 0x{:04x}. Skipping {} bytes".format( + h.type, h.size + ) + ) self.buff.seek(h.end) continue # Check that we read a correct header if h.header_size != 0x10: - logger.error("XML Resource Type Chunk header size does not match 16! " - "At chunk type 0x{:04x}, declared header size=0x{:04x}, chunk size=0x{:04x}".format(h.type, h.header_size, h.size)) + logger.error( + "XML Resource Type Chunk header size does not match 16! " + "At chunk type 0x{:04x}, declared header size=0x{:04x}, chunk size=0x{:04x}".format( + h.type, h.header_size, h.size + ) + ) self.buff.seek(h.end) continue # Line Number of the source file, only used as meta information - self.m_lineNumber, = unpack(' uri {}: '{}'".format(prefix, s_prefix, uri, s_uri)) + "Start of Namespace mapping: prefix {}: '{}' --> uri {}: '{}'".format( + prefix, s_prefix, uri, s_uri + ) + ) if s_uri == '': - logger.warning("Namespace prefix '{}' resolves to empty URI. " - "This might be a packer.".format(s_prefix)) + logger.warning( + "Namespace prefix '{}' resolves to empty URI. " + "This might be a packer.".format(s_prefix) + ) if (prefix, uri) in self.namespaces: - logger.debug("Namespace mapping ({}, {}) already seen! " - "This is usually not a problem but could indicate packers or broken AXML compilers.".format(prefix, uri)) + logger.debug( + "Namespace mapping ({}, {}) already seen! " + "This is usually not a problem but could indicate packers or broken AXML compilers.".format( + prefix, uri + ) + ) self.namespaces.append((prefix, uri)) # We can continue with the next chunk, as we store the namespace @@ -590,15 +707,17 @@ class AXMLParser: if h.type == RES_XML_END_NAMESPACE_TYPE: # END_PREFIX contains again prefix and uri field - prefix, = unpack('> 16) - 1 self.m_attribute_count = attributeCount & 0xFFFF @@ -641,20 +760,26 @@ class AXMLParser: # * Type # * Data for j in range(0, ATTRIBUTE_LENGTH): - self.m_attributes.append(unpack('> 24 self.m_event = START_TAG break if h.type == RES_XML_END_ELEMENT_TYPE: - self.m_namespaceUri, = unpack(' uint32_t index - self.m_name, = unpack(' str: """ - Return the String assosciated with the tag name + Return the String associated with the tag name """ - if self.m_name == -1 or (self.m_event != START_TAG and self.m_event != END_TAG): + if self.m_name == -1 or ( + self.m_event != START_TAG and self.m_event != END_TAG + ): return '' return self.sb[self.m_name] @property - def comment(self): + def comment(self) -> Union[str, None]: """ Return the comment at the current position or None if no comment is given @@ -715,11 +845,13 @@ class AXMLParser: return self.sb[self.m_comment_index] @property - def namespace(self): + def namespace(self) -> str: """ Return the Namespace URI (if any) as a String for the current tag """ - if self.m_name == -1 or (self.m_event != START_TAG and self.m_event != END_TAG): + if self.m_name == -1 or ( + self.m_event != START_TAG and self.m_event != END_TAG + ): return '' # No Namespace @@ -729,7 +861,7 @@ class AXMLParser: return self.sb[self.m_namespaceUri] @property - def nsmap(self): + def nsmap(self) -> dict[str, str]: """ Returns the current namespace mapping as a dictionary @@ -750,12 +882,12 @@ class AXMLParser: # Solve 2) & 4) by not including if s_uri != "" and s_prefix != "": # solve 1) by using the last one in the list - NSMAP[s_prefix] = s_uri + NSMAP[s_prefix] = s_uri.strip() return NSMAP @property - def text(self): + def text(self) -> str: """ Return the String assosicated with the current text """ @@ -764,21 +896,21 @@ class AXMLParser: return self.sb[self.m_name] - def getName(self): + def getName(self) -> str: """ Legacy only! use :py:attr:`~androguard.core.bytecodes.AXMLParser.name` instead """ return self.name - def getText(self): + def getText(self) -> str: """ Legacy only! use :py:attr:`~androguard.core.bytecodes.AXMLParser.text` instead """ return self.text - def getPrefix(self): + def getPrefix(self) -> str: """ Legacy only! use :py:attr:`~androguard.core.bytecodes.AXMLParser.namespace` instead @@ -798,7 +930,7 @@ class AXMLParser: return offset - def getAttributeCount(self): + def getAttributeCount(self) -> int: """ Return the number of Attributes for a Tag or -1 if not in a tag @@ -808,7 +940,7 @@ class AXMLParser: return self.m_attribute_count - def getAttributeUri(self, index): + def getAttributeUri(self, index: int): """ Returns the numeric ID for the namespace URI of an attribute """ @@ -819,7 +951,7 @@ class AXMLParser: return uri - def getAttributeNamespace(self, index): + def getAttributeNamespace(self, index: int): """ Return the Namespace URI (if any) for the attribute """ @@ -833,7 +965,7 @@ class AXMLParser: return self.sb[uri] - def getAttributeName(self, index): + def getAttributeName(self, index: int): """ Returns the String which represents the attribute name """ @@ -843,18 +975,22 @@ class AXMLParser: res = self.sb[name] # If the result is a (null) string, we need to look it up. - if not res or res == ":": + if name < len(self.m_resourceIDs): attr = self.m_resourceIDs[name] if attr in public.SYSTEM_RESOURCES['attributes']['inverse']: - res = 'android:' + \ - public.SYSTEM_RESOURCES['attributes']['inverse'][attr] - else: - # Attach the HEX Number, so for multiple missing attributes we do not run - # into problems. - res = 'android:UNKNOWN_SYSTEM_ATTRIBUTE_{:08x}'.format(attr) + res = public.SYSTEM_RESOURCES['attributes']['inverse'][ + attr + ].replace("_", ":") + if res != self.sb[name]: + self.packerwarning = True + + if not res or res == ":": + # Attach the HEX Number, so for multiple missing attributes we do not run + # into problems. + res = 'android:UNKNOWN_SYSTEM_ATTRIBUTE_{:08x}'.format(attr) return res - def getAttributeValueType(self, index): + def getAttributeValueType(self, index: int): """ Return the type of the attribute at the given index @@ -865,7 +1001,7 @@ class AXMLParser: offset = self._get_attribute_offset(index) return self.m_attributes[offset + ATTRIBUTE_IX_VALUE_TYPE] - def getAttributeValueData(self, index): + def getAttributeValueData(self, index: int): """ Return the data of the attribute at the given index @@ -895,7 +1031,9 @@ class AXMLParser: return '' -def format_value(_type, _data, lookup_string=lambda ix: ""): +def format_value( + _type: int, _data: int, lookup_string=lambda ix: "" +) -> str: """ Format a value based on type and data. By default, no strings are looked up and "" is returned. @@ -935,10 +1073,15 @@ def format_value(_type, _data, lookup_string=lambda ix: ""): return "true" elif _type == TYPE_DIMENSION: - return "{:f}{}".format(complexToFloat(_data), DIMENSION_UNITS[_data & COMPLEX_UNIT_MASK]) + return "{:f}{}".format( + complexToFloat(_data), DIMENSION_UNITS[_data & COMPLEX_UNIT_MASK] + ) elif _type == TYPE_FRACTION: - return "{:f}{}".format(complexToFloat(_data) * 100, FRACTION_UNITS[_data & COMPLEX_UNIT_MASK]) + return "{:f}{}".format( + complexToFloat(_data) * 100, + FRACTION_UNITS[_data & COMPLEX_UNIT_MASK], + ) elif TYPE_FIRST_COLOR_INT <= _type <= TYPE_LAST_COLOR_INT: return "#%08X" % _data @@ -956,10 +1099,11 @@ class AXMLPrinter: A Reference Implementation can be found at http://androidxref.com/9.0.0_r3/xref/frameworks/base/tools/aapt/XMLNode.cpp """ + __charrange = None __replacement = None - def __init__(self, raw_buff): + def __init__(self, raw_buff: bytes) -> bytes: logger.debug("AXMLPrinter") self.axml = AXMLParser(raw_buff) @@ -973,6 +1117,9 @@ class AXMLPrinter: logger.debug("DEBUG ARSC TYPE {}".format(_type)) if _type == START_TAG: + if not self.axml.name: # Check if the name is empty + logger.debug("Empty tag name, skipping to next element") + continue # Skip this iteration uri = self._print_namespace(self.axml.namespace) uri, name = self._fix_name(uri, self.axml.name) tag = "{}{}".format(uri, name) @@ -981,24 +1128,52 @@ class AXMLPrinter: if comment: if self.root is None: logger.warning( - "Can not attach comment with content '{}' without root!".format(comment)) + "Can not attach comment with content '{}' without root!".format( + comment + ) + ) else: cur[-1].append(etree.Comment(comment)) - logger.debug("START_TAG: {} (line={})".format( - tag, self.axml.m_lineNumber)) - elem = etree.Element(tag, nsmap=self.axml.nsmap) + logger.debug( + "START_TAG: {} (line={})".format( + tag, self.axml.m_lineNumber + ) + ) + + try: + elem = etree.Element(tag, nsmap=self.axml.nsmap) + except ValueError as e: + logger.error(e) + # nsmap= {'