releasetools: Clean up check_target_files_signatures.py.

And make it Python 3 compatibile.

Bug: 131631303
Test: python check_target_files_signatures.py target-files.zip
Test: python3 check_target_files_signatures.py target-files.zip
Change-Id: I38e868c3124387f5543ad68c92c75917e40b7d8f
This commit is contained in:
Tao Bao 2017-12-07 10:33:34 -08:00
parent 14c01bea97
commit a67e12d66f
1 changed files with 63 additions and 54 deletions

View File

@ -39,8 +39,11 @@ Usage: check_target_file_signatures [flags] target_files
"""
from __future__ import print_function
import logging
import os
import os.path
import re
import subprocess
import sys
@ -49,7 +52,7 @@ import zipfile
import common
if sys.hexversion < 0x02070000:
print >> sys.stderr, "Python 2.7 or newer is required."
print("Python 2.7 or newer is required.", file=sys.stderr)
sys.exit(1)
@ -65,8 +68,10 @@ logger = logging.getLogger(__name__)
class MyZipInfo(zipfile.ZipInfo):
def _decodeExtra(self):
pass
zipfile.ZipInfo = MyZipInfo
OPTIONS = common.OPTIONS
OPTIONS.text = False
@ -76,28 +81,34 @@ OPTIONS.local_cert_dirs = ("vendor", "build")
PROBLEMS = []
PROBLEM_PREFIX = []
def AddProblem(msg):
PROBLEMS.append(" ".join(PROBLEM_PREFIX) + " " + msg)
def Push(msg):
PROBLEM_PREFIX.append(msg)
def Pop():
PROBLEM_PREFIX.pop()
def Banner(msg):
print "-" * 70
print " ", msg
print "-" * 70
print("-" * 70)
print(" ", msg)
print("-" * 70)
def GetCertSubject(cert):
p = common.Run(["openssl", "x509", "-inform", "DER", "-text"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE)
stdout=subprocess.PIPE,
universal_newlines=False)
out, err = p.communicate(cert)
if err and not err.strip():
return "(error reading cert subject)"
for line in out.split("\n"):
for line in out.decode().split("\n"):
line = line.strip()
if line.startswith("Subject:"):
return line[8:].strip()
@ -105,6 +116,7 @@ def GetCertSubject(cert):
class CertDB(object):
def __init__(self):
self.certs = {}
@ -132,13 +144,13 @@ class CertDB(object):
to_load.extend(certs)
for i in to_load:
f = open(i)
cert = common.ParseCertificate(f.read())
f.close()
with open(i) as f:
cert = common.ParseCertificate(f.read())
name, _ = os.path.splitext(i)
name, _ = os.path.splitext(name)
self.Add(cert, name)
ALL_CERTS = CertDB()
@ -152,13 +164,14 @@ def CertFromPKCS7(data, filename):
"-outform", "PEM",
"-print_certs"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE)
stdout=subprocess.PIPE,
universal_newlines=False)
out, err = p.communicate(data)
if err and not err.strip():
AddProblem("error reading cert:\n" + err)
AddProblem("error reading cert:\n" + err.decode())
return None
cert = common.ParseCertificate(out)
cert = common.ParseCertificate(out.decode())
if not cert:
AddProblem("error parsing cert output")
return None
@ -184,22 +197,20 @@ class APK(object):
def RecordCerts(self, full_filename):
out = set()
try:
f = open(full_filename)
apk = zipfile.ZipFile(f, "r")
with zipfile.ZipFile(full_filename) as apk:
pkcs7 = None
for info in apk.infolist():
if info.filename.startswith("META-INF/") and \
(info.filename.endswith(".DSA") or info.filename.endswith(".RSA")):
pkcs7 = apk.read(info.filename)
cert = CertFromPKCS7(pkcs7, info.filename)
filename = info.filename
if (filename.startswith("META-INF/") and
info.filename.endswith((".DSA", ".RSA"))):
pkcs7 = apk.read(filename)
cert = CertFromPKCS7(pkcs7, filename)
out.add(cert)
ALL_CERTS.Add(cert)
if not pkcs7:
AddProblem("no signature")
finally:
f.close()
self.certs = frozenset(out)
self.certs = frozenset(out)
def ReadManifest(self, full_filename):
p = common.Run(["aapt", "dump", "xmltree", full_filename,
@ -247,8 +258,8 @@ class TargetFiles(object):
# This is the list of wildcards of files we extract from |filename|.
apk_extensions = ['*.apk', '*.apex']
self.certmap, compressed_extension = common.ReadApkCerts(
zipfile.ZipFile(filename))
with zipfile.ZipFile(filename) as input_zip:
self.certmap, compressed_extension = common.ReadApkCerts(input_zip)
if compressed_extension:
apk_extensions.append('*.apk' + compressed_extension)
@ -287,7 +298,7 @@ class TargetFiles(object):
"""Look for any instances where packages signed with different
certs request the same sharedUserId."""
apks_by_uid = {}
for apk in self.apks.itervalues():
for apk in self.apks.values():
if apk.shared_uid:
apks_by_uid.setdefault(apk.shared_uid, []).append(apk)
@ -302,15 +313,15 @@ class TargetFiles(object):
AddProblem("different cert sets for packages with uid %s" % (uid,))
print "uid %s is shared by packages with different cert sets:" % (uid,)
print("uid %s is shared by packages with different cert sets:" % (uid,))
for apk in apks:
print "%-*s [%s]" % (self.max_pkg_len, apk.package, apk.filename)
print("%-*s [%s]" % (self.max_pkg_len, apk.package, apk.filename))
for cert in apk.certs:
print " ", ALL_CERTS.Get(cert)
print
print(" ", ALL_CERTS.Get(cert))
print()
def CheckExternalSignatures(self):
for apk_filename, certname in self.certmap.iteritems():
for apk_filename, certname in self.certmap.items():
if certname == "EXTERNAL":
# Apps marked EXTERNAL should be signed with the test key
# during development, then manually re-signed after
@ -326,25 +337,25 @@ class TargetFiles(object):
def PrintCerts(self):
"""Display a table of packages grouped by cert."""
by_cert = {}
for apk in self.apks.itervalues():
for apk in self.apks.values():
for cert in apk.certs:
by_cert.setdefault(cert, []).append((apk.package, apk))
order = [(-len(v), k) for (k, v) in by_cert.iteritems()]
order = [(-len(v), k) for (k, v) in by_cert.items()]
order.sort()
for _, cert in order:
print "%s:" % (ALL_CERTS.Get(cert),)
print("%s:" % (ALL_CERTS.Get(cert),))
apks = by_cert[cert]
apks.sort()
for _, apk in apks:
if apk.shared_uid:
print " %-*s %-*s [%s]" % (self.max_fn_len, apk.filename,
print(" %-*s %-*s [%s]" % (self.max_fn_len, apk.filename,
self.max_pkg_len, apk.package,
apk.shared_uid)
apk.shared_uid))
else:
print " %-*s %s" % (self.max_fn_len, apk.filename, apk.package)
print
print(" %-*s %s" % (self.max_fn_len, apk.filename, apk.package))
print()
def CompareWith(self, other):
"""Look for instances where a given package that exists in both
@ -365,12 +376,12 @@ class TargetFiles(object):
by_certpair.setdefault((other.apks[i].certs,
self.apks[i].certs), []).append(i)
else:
print "%s [%s]: new APK (not in comparison target_files)" % (
i, self.apks[i].filename)
print("%s [%s]: new APK (not in comparison target_files)" % (
i, self.apks[i].filename))
else:
if i in other.apks:
print "%s [%s]: removed APK (only in comparison target_files)" % (
i, other.apks[i].filename)
print("%s [%s]: removed APK (only in comparison target_files)" % (
i, other.apks[i].filename))
if by_certpair:
AddProblem("some APKs changed certs")
@ -378,23 +389,23 @@ class TargetFiles(object):
for (old, new), packages in sorted(by_certpair.items()):
for i, o in enumerate(old):
if i == 0:
print "was", ALL_CERTS.Get(o)
print("was", ALL_CERTS.Get(o))
else:
print " ", ALL_CERTS.Get(o)
print(" ", ALL_CERTS.Get(o))
for i, n in enumerate(new):
if i == 0:
print "now", ALL_CERTS.Get(n)
print("now", ALL_CERTS.Get(n))
else:
print " ", ALL_CERTS.Get(n)
print(" ", ALL_CERTS.Get(n))
for i in sorted(packages):
old_fn = other.apks[i].filename
new_fn = self.apks[i].filename
if old_fn == new_fn:
print " %-*s [%s]" % (max_pkg_len, i, old_fn)
print(" %-*s [%s]" % (max_pkg_len, i, old_fn))
else:
print " %-*s [was: %s; now: %s]" % (max_pkg_len, i,
old_fn, new_fn)
print
print(" %-*s [was: %s; now: %s]" % (max_pkg_len, i,
old_fn, new_fn))
print()
def main(argv):
@ -451,9 +462,9 @@ def main(argv):
target_files.CompareWith(compare_files)
if PROBLEMS:
print "%d problem(s) found:\n" % (len(PROBLEMS),)
print("%d problem(s) found:\n" % (len(PROBLEMS),))
for p in PROBLEMS:
print p
print(p)
return 1
return 0
@ -464,9 +475,7 @@ if __name__ == '__main__':
r = main(sys.argv[1:])
sys.exit(r)
except common.ExternalError as e:
print
print " ERROR: %s" % (e,)
print
print("\n ERROR: %s\n" % (e,))
sys.exit(1)
finally:
common.Cleanup()