dba59eeabd
In addition to the unzipping work, common.UnzipTemp() kindly bundles an open ZipFile object as part of the return value. It doesn't look very helpful to the callers though. It also looks less obvious that the caller needs to properly close the handle (missing the close here is benign though). This CL just removes the ZipFile object out of the return value, and leaves the work to callers. Test: `m dist` on both of A/B and non-A/B target. Test: python -m unittest test_add_img_to_target_files Test: python -m unittest test_common Test: python -m unittest test_ota_from_target_files Test: Check the callers to common.UnzipTemp() in code search. Change-Id: Id47da3fd42a0e76d6ae8851f05780db319ee48cf
469 lines
14 KiB
Python
Executable File
469 lines
14 KiB
Python
Executable File
#!/usr/bin/env python
|
|
#
|
|
# Copyright (C) 2009 The Android Open Source Project
|
|
#
|
|
# 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
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# 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.
|
|
|
|
"""
|
|
Check the signatures of all APKs in a target_files .zip file. With
|
|
-c, compare the signatures of each package to the ones in a separate
|
|
target_files (usually a previously distributed build for the same
|
|
device) and flag any changes.
|
|
|
|
Usage: check_target_file_signatures [flags] target_files
|
|
|
|
-c (--compare_with) <other_target_files>
|
|
Look for compatibility problems between the two sets of target
|
|
files (eg., packages whose keys have changed).
|
|
|
|
-l (--local_cert_dirs) <dir,dir,...>
|
|
Comma-separated list of top-level directories to scan for
|
|
.x509.pem files. Defaults to "vendor,build". Where cert files
|
|
can be found that match APK signatures, the filename will be
|
|
printed as the cert name, otherwise a hash of the cert plus its
|
|
subject string will be printed instead.
|
|
|
|
-t (--text)
|
|
Dump the certificate information for both packages in comparison
|
|
mode (this output is normally suppressed).
|
|
|
|
"""
|
|
|
|
import sys
|
|
|
|
if sys.hexversion < 0x02070000:
|
|
print >> sys.stderr, "Python 2.7 or newer is required."
|
|
sys.exit(1)
|
|
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import zipfile
|
|
|
|
import common
|
|
|
|
# Work around a bug in Python's zipfile module that prevents opening of zipfiles
|
|
# if any entry has an extra field of between 1 and 3 bytes (which is common with
|
|
# zipaligned APKs). This overrides the ZipInfo._decodeExtra() method (which
|
|
# contains the bug) with an empty version (since we don't need to decode the
|
|
# extra field anyway).
|
|
# Issue #14315: https://bugs.python.org/issue14315, fixed in Python 2.7.8 and
|
|
# Python 3.5.0 alpha 1.
|
|
class MyZipInfo(zipfile.ZipInfo):
|
|
def _decodeExtra(self):
|
|
pass
|
|
zipfile.ZipInfo = MyZipInfo
|
|
|
|
OPTIONS = common.OPTIONS
|
|
|
|
OPTIONS.text = False
|
|
OPTIONS.compare_with = None
|
|
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
|
|
|
|
|
|
def GetCertSubject(cert):
|
|
p = common.Run(["openssl", "x509", "-inform", "DER", "-text"],
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE)
|
|
out, err = p.communicate(cert)
|
|
if err and not err.strip():
|
|
return "(error reading cert subject)"
|
|
for line in out.split("\n"):
|
|
line = line.strip()
|
|
if line.startswith("Subject:"):
|
|
return line[8:].strip()
|
|
return "(unknown cert subject)"
|
|
|
|
|
|
class CertDB(object):
|
|
def __init__(self):
|
|
self.certs = {}
|
|
|
|
def Add(self, cert, name=None):
|
|
if cert in self.certs:
|
|
if name:
|
|
self.certs[cert] = self.certs[cert] + "," + name
|
|
else:
|
|
if name is None:
|
|
name = "unknown cert %s (%s)" % (common.sha1(cert).hexdigest()[:12],
|
|
GetCertSubject(cert))
|
|
self.certs[cert] = name
|
|
|
|
def Get(self, cert):
|
|
"""Return the name for a given cert."""
|
|
return self.certs.get(cert, None)
|
|
|
|
def FindLocalCerts(self):
|
|
to_load = []
|
|
for top in OPTIONS.local_cert_dirs:
|
|
for dirpath, _, filenames in os.walk(top):
|
|
certs = [os.path.join(dirpath, i)
|
|
for i in filenames if i.endswith(".x509.pem")]
|
|
if certs:
|
|
to_load.extend(certs)
|
|
|
|
for i in to_load:
|
|
f = open(i)
|
|
cert = common.ParseCertificate(f.read())
|
|
f.close()
|
|
name, _ = os.path.splitext(i)
|
|
name, _ = os.path.splitext(name)
|
|
self.Add(cert, name)
|
|
|
|
ALL_CERTS = CertDB()
|
|
|
|
|
|
def CertFromPKCS7(data, filename):
|
|
"""Read the cert out of a PKCS#7-format file (which is what is
|
|
stored in a signed .apk)."""
|
|
Push(filename + ":")
|
|
try:
|
|
p = common.Run(["openssl", "pkcs7",
|
|
"-inform", "DER",
|
|
"-outform", "PEM",
|
|
"-print_certs"],
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE)
|
|
out, err = p.communicate(data)
|
|
if err and not err.strip():
|
|
AddProblem("error reading cert:\n" + err)
|
|
return None
|
|
|
|
cert = common.ParseCertificate(out)
|
|
if not cert:
|
|
AddProblem("error parsing cert output")
|
|
return None
|
|
return cert
|
|
finally:
|
|
Pop()
|
|
|
|
|
|
class APK(object):
|
|
def __init__(self, full_filename, filename):
|
|
self.filename = filename
|
|
self.certs = None
|
|
self.shared_uid = None
|
|
self.package = None
|
|
|
|
Push(filename+":")
|
|
try:
|
|
self.RecordCerts(full_filename)
|
|
self.ReadManifest(full_filename)
|
|
finally:
|
|
Pop()
|
|
|
|
def RecordCerts(self, full_filename):
|
|
out = set()
|
|
try:
|
|
f = open(full_filename)
|
|
apk = zipfile.ZipFile(f, "r")
|
|
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)
|
|
out.add(cert)
|
|
ALL_CERTS.Add(cert)
|
|
if not pkcs7:
|
|
AddProblem("no signature")
|
|
finally:
|
|
f.close()
|
|
self.certs = frozenset(out)
|
|
|
|
def ReadManifest(self, full_filename):
|
|
p = common.Run(["aapt", "dump", "xmltree", full_filename,
|
|
"AndroidManifest.xml"],
|
|
stdout=subprocess.PIPE)
|
|
manifest, err = p.communicate()
|
|
if err:
|
|
AddProblem("failed to read manifest")
|
|
return
|
|
|
|
self.shared_uid = None
|
|
self.package = None
|
|
|
|
for line in manifest.split("\n"):
|
|
line = line.strip()
|
|
m = re.search(r'A: (\S*?)(?:\(0x[0-9a-f]+\))?="(.*?)" \(Raw', line)
|
|
if m:
|
|
name = m.group(1)
|
|
if name == "android:sharedUserId":
|
|
if self.shared_uid is not None:
|
|
AddProblem("multiple sharedUserId declarations")
|
|
self.shared_uid = m.group(2)
|
|
elif name == "package":
|
|
if self.package is not None:
|
|
AddProblem("multiple package declarations")
|
|
self.package = m.group(2)
|
|
|
|
if self.package is None:
|
|
AddProblem("no package declaration")
|
|
|
|
|
|
class TargetFiles(object):
|
|
def __init__(self):
|
|
self.max_pkg_len = 30
|
|
self.max_fn_len = 20
|
|
self.apks = None
|
|
self.apks_by_basename = None
|
|
self.certmap = None
|
|
|
|
def LoadZipFile(self, filename):
|
|
# First read the APK certs file to figure out whether there are compressed
|
|
# APKs in the archive. If we do have compressed APKs in the archive, then we
|
|
# must decompress them individually before we perform any analysis.
|
|
|
|
# This is the list of wildcards of files we extract from |filename|.
|
|
apk_extensions = ['*.apk']
|
|
|
|
self.certmap, compressed_extension = common.ReadApkCerts(zipfile.ZipFile(filename, "r"))
|
|
if compressed_extension:
|
|
apk_extensions.append("*.apk" + compressed_extension)
|
|
|
|
d = common.UnzipTemp(filename, apk_extensions)
|
|
try:
|
|
self.apks = {}
|
|
self.apks_by_basename = {}
|
|
for dirpath, _, filenames in os.walk(d):
|
|
for fn in filenames:
|
|
# Decompress compressed APKs before we begin processing them.
|
|
if compressed_extension and fn.endswith(compressed_extension):
|
|
# First strip the compressed extension from the file.
|
|
uncompressed_fn = fn[:-len(compressed_extension)]
|
|
|
|
# Decompress the compressed file to the output file.
|
|
common.Gunzip(os.path.join(dirpath, fn),
|
|
os.path.join(dirpath, uncompressed_fn))
|
|
|
|
# Finally, delete the compressed file and use the uncompressed file
|
|
# for further processing. Note that the deletion is not strictly required,
|
|
# but is done here to ensure that we're not using too much space in
|
|
# the temporary directory.
|
|
os.remove(os.path.join(dirpath, fn))
|
|
fn = uncompressed_fn
|
|
|
|
|
|
if fn.endswith(".apk"):
|
|
fullname = os.path.join(dirpath, fn)
|
|
displayname = fullname[len(d)+1:]
|
|
apk = APK(fullname, displayname)
|
|
self.apks[apk.filename] = apk
|
|
self.apks_by_basename[os.path.basename(apk.filename)] = apk
|
|
|
|
self.max_pkg_len = max(self.max_pkg_len, len(apk.package))
|
|
self.max_fn_len = max(self.max_fn_len, len(apk.filename))
|
|
finally:
|
|
shutil.rmtree(d)
|
|
|
|
def CheckSharedUids(self):
|
|
"""Look for any instances where packages signed with different
|
|
certs request the same sharedUserId."""
|
|
apks_by_uid = {}
|
|
for apk in self.apks.itervalues():
|
|
if apk.shared_uid:
|
|
apks_by_uid.setdefault(apk.shared_uid, []).append(apk)
|
|
|
|
for uid in sorted(apks_by_uid.keys()):
|
|
apks = apks_by_uid[uid]
|
|
for apk in apks[1:]:
|
|
if apk.certs != apks[0].certs:
|
|
break
|
|
else:
|
|
# all packages have the same set of certs; this uid is fine.
|
|
continue
|
|
|
|
AddProblem("different cert sets for packages with uid %s" % (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)
|
|
for cert in apk.certs:
|
|
print " ", ALL_CERTS.Get(cert)
|
|
print
|
|
|
|
def CheckExternalSignatures(self):
|
|
for apk_filename, certname in self.certmap.iteritems():
|
|
if certname == "EXTERNAL":
|
|
# Apps marked EXTERNAL should be signed with the test key
|
|
# during development, then manually re-signed after
|
|
# predexopting. Consider it an error if this app is now
|
|
# signed with any key that is present in our tree.
|
|
apk = self.apks_by_basename[apk_filename]
|
|
name = ALL_CERTS.Get(apk.cert)
|
|
if not name.startswith("unknown "):
|
|
Push(apk.filename)
|
|
AddProblem("hasn't been signed with EXTERNAL cert")
|
|
Pop()
|
|
|
|
def PrintCerts(self):
|
|
"""Display a table of packages grouped by cert."""
|
|
by_cert = {}
|
|
for apk in self.apks.itervalues():
|
|
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.sort()
|
|
|
|
for _, cert in order:
|
|
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,
|
|
self.max_pkg_len, apk.package,
|
|
apk.shared_uid)
|
|
else:
|
|
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
|
|
self and other have different certs."""
|
|
|
|
all_apks = set(self.apks.keys())
|
|
all_apks.update(other.apks.keys())
|
|
|
|
max_pkg_len = max(self.max_pkg_len, other.max_pkg_len)
|
|
|
|
by_certpair = {}
|
|
|
|
for i in all_apks:
|
|
if i in self.apks:
|
|
if i in other.apks:
|
|
# in both; should have same set of certs
|
|
if self.apks[i].certs != other.apks[i].certs:
|
|
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)
|
|
else:
|
|
if i in other.apks:
|
|
print "%s [%s]: removed APK (only in comparison target_files)" % (
|
|
i, other.apks[i].filename)
|
|
|
|
if by_certpair:
|
|
AddProblem("some APKs changed certs")
|
|
Banner("APK signing differences")
|
|
for (old, new), packages in sorted(by_certpair.items()):
|
|
for i, o in enumerate(old):
|
|
if i == 0:
|
|
print "was", ALL_CERTS.Get(o)
|
|
else:
|
|
print " ", ALL_CERTS.Get(o)
|
|
for i, n in enumerate(new):
|
|
if i == 0:
|
|
print "now", ALL_CERTS.Get(n)
|
|
else:
|
|
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)
|
|
else:
|
|
print " %-*s [was: %s; now: %s]" % (max_pkg_len, i,
|
|
old_fn, new_fn)
|
|
print
|
|
|
|
|
|
def main(argv):
|
|
def option_handler(o, a):
|
|
if o in ("-c", "--compare_with"):
|
|
OPTIONS.compare_with = a
|
|
elif o in ("-l", "--local_cert_dirs"):
|
|
OPTIONS.local_cert_dirs = [i.strip() for i in a.split(",")]
|
|
elif o in ("-t", "--text"):
|
|
OPTIONS.text = True
|
|
else:
|
|
return False
|
|
return True
|
|
|
|
args = common.ParseOptions(argv, __doc__,
|
|
extra_opts="c:l:t",
|
|
extra_long_opts=["compare_with=",
|
|
"local_cert_dirs="],
|
|
extra_option_handler=option_handler)
|
|
|
|
if len(args) != 1:
|
|
common.Usage(__doc__)
|
|
sys.exit(1)
|
|
|
|
ALL_CERTS.FindLocalCerts()
|
|
|
|
Push("input target_files:")
|
|
try:
|
|
target_files = TargetFiles()
|
|
target_files.LoadZipFile(args[0])
|
|
finally:
|
|
Pop()
|
|
|
|
compare_files = None
|
|
if OPTIONS.compare_with:
|
|
Push("comparison target_files:")
|
|
try:
|
|
compare_files = TargetFiles()
|
|
compare_files.LoadZipFile(OPTIONS.compare_with)
|
|
finally:
|
|
Pop()
|
|
|
|
if OPTIONS.text or not compare_files:
|
|
Banner("target files")
|
|
target_files.PrintCerts()
|
|
target_files.CheckSharedUids()
|
|
target_files.CheckExternalSignatures()
|
|
if compare_files:
|
|
if OPTIONS.text:
|
|
Banner("comparison files")
|
|
compare_files.PrintCerts()
|
|
target_files.CompareWith(compare_files)
|
|
|
|
if PROBLEMS:
|
|
print "%d problem(s) found:\n" % (len(PROBLEMS),)
|
|
for p in PROBLEMS:
|
|
print p
|
|
return 1
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
r = main(sys.argv[1:])
|
|
sys.exit(r)
|
|
except common.ExternalError as e:
|
|
print
|
|
print " ERROR: %s" % (e,)
|
|
print
|
|
sys.exit(1)
|