blob: 6dbbeaf91571fc938abea481c645480a995576bc [file] [log] [blame]
#!/usr/bin/env python3
# Copyright 2019, 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
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.
import argparse
import asyncio
import csv
import itertools
import os
import re
import struct
import sys
import tempfile
import time
import zipfile
from typing import Any, Callable, Dict, Generic, Iterable, List, NamedTuple, TextIO, Tuple, TypeVar, Optional, Union
# Include generated protos.
dir_name = os.path.dirname(os.path.realpath(__file__))
sys.path.append(dir_name + "/generated")
from TraceFile_pb2 import *
def parse_options(argv: List[str] = None):
"""Parse command line arguments and return an argparse Namespace object."""
parser = argparse.ArgumentParser(description="Compile a TraceFile.proto from a manual text file.")
# argparse considers args starting with - and -- optional in --help, even though required=True.
# by using a named argument group --help will clearly say that it's required instead of optional.
required_named = parser.add_argument_group('required named arguments')
# optional arguments
# use a group here to get the required arguments to appear 'above' the optional arguments in help.
optional_named = parser.add_argument_group('optional named arguments')
optional_named.add_argument('-opb', '--output-proto-binary', dest='output_proto_binary', action='store', help='Write binary proto output to file.')
optional_named.add_argument('-pm', '--pinlist-meta', dest='pinlist_meta', action='store', help='Path to pinlist.meta (default=none) binary file.')
optional_named.add_argument('-pmp', '--pinlist-meta-parent', dest='pinlist_meta_parent', action='store', help='Device path that the pinlist.meta applies to (e.g. /data/.../somefile.apk)')
optional_named.add_argument('-i', '--input', dest='input', action='store', help='Input text file (default stdin).')
optional_named.add_argument('-zp', '--zip_path', dest='zip_path', action='append', help='Directory containing zip files.')
optional_named.add_argument('-d', '--debug', dest='debug', action='store_true', help='Add extra debugging output')
optional_named.add_argument('-ot', '--output-text', dest='output_text', action='store', help='Output text file (default stdout).')
return parser.parse_args(argv)
# TODO: refactor this with a common library file with
def _debug_print(*args, **kwargs):
"""Print the args to sys.stderr if the --debug/-d flag was passed in."""
if _debug:
print(*args, **kwargs, file=sys.stderr)
class BadInputError(Exception):
InputRecord = NamedTuple('InputRecord', [('filepath', str), ('offset', int), ('length', int), ('remark', str)])
def find_zip_in_paths(original_name, zip_paths):
# /foo/bar/ ->
file_basename = os.path.split(original_name)[1]
# the file must be located in one of the --zip-path arguments
matched = None
for zip_path in zip_paths:
for dir_entry in os.listdir(zip_path):
if dir_entry == file_basename:
matched = os.path.join(zip_path, dir_entry)
if matched:
if not matched:
raise ValueError("%s could not be found in any of the --zip_path specified." %(file_basename))
_debug_print("found zip file ", file_basename, " in ", matched)
if not zipfile.is_zipfile(matched):
raise ValueError("%s is not a zip file" %(matched))
return matched
def handle_zip_entry(input_record, zip_paths):
res = re.match("([^!]+)[!](.*)", input_record.filepath)
if not res:
return input_record
# 'foo!bar'
in_filepath = res[1] # -> 'foo'
in_zip_entry = res[2] # -> 'bar'
matched = find_zip_in_paths(in_filepath, zip_paths)
zip = zipfile.ZipFile(matched)
zip_info = zip.getinfo(in_zip_entry)
except KeyError:
raise ValueError("%s is not an item in the zip file %s" %(in_zip_entry, matched))
# TODO: do we also need to add header size to this?
in_offset = zip_info.header_offset
# TODO: if a range is specified, use that instead.
in_length = zip_info.compress_size
return InputRecord(in_filepath, in_offset, in_length, 'zip entry (%s)' %(in_zip_entry))
def parse_input_file(input: Iterable[str], zip_paths: List[str]) -> Iterable[InputRecord]:
for line in input:
line = line.strip()
_debug_print("Line = ", line)
if not line:
_debug_print(" skip empty line", line)
elif line[0] == "#":
_debug_print(" skip commented line", line)
res = re.match("([^\s]+)\s+(\d+)\s+(\d+)", line)
if not res:
raise BadInputError("Expected input of form: <str:filepath> <int:offset> <int:length>")
in_filepath = res[1]
in_offset = int(res[2])
in_length = int(res[3])
yield handle_zip_entry(InputRecord(in_filepath, in_offset, in_length, 'regular file'), zip_paths)
# format:
# (<big_endian(i32):file_offset> <big_endian(i32):range_length>)+
def parse_pin_meta(pin_meta_file, pinlist_meta_parent, zip_paths):
if not pin_meta_file:
return ()
# '/data/app/'
# -> ''
package_name_match = re.match('/.*/(.*)-.*=/base.apk', pinlist_meta_parent)
if not package_name_match:
raise ValueError("%s did not contain the <packagename>.apk" %(pinlist_meta_parent))
package_name = package_name_match[1]
# "" -> "GoogleCamera.apk"
apk_name = package_name.split(".")[-1] + ".apk"
path_to_zip_on_host = find_zip_in_paths(apk_name, zip_paths)
apk_file_size = os.path.getsize(path_to_zip_on_host)
_debug_print("APK path '%s' file size '%d'" %(path_to_zip_on_host, apk_file_size))
while True:
data =
if not data:
(pin_offset, pin_length) = struct.unpack(PIN_META_FORMAT, data) # (offset, length)
remark = 'regular file (pinlist.meta)'
remaining_size = apk_file_size - pin_offset
if remaining_size < 0:
print("WARNING: Clamp entry (%d, %d), offset too large (max file size = %d)" %(pin_offset, pin_length, apk_file_size))
pin_length = pin_length + remaining_size
pin_offset = pin_offset + remaining_size
if pin_offset < 0:
pin_offset = 0
remark += '[clamped.offset]'
pin_last_offset = pin_offset + pin_length
remaining_size = apk_file_size - pin_last_offset
if remaining_size < 0:
print("WARNING: Clamp entry (%d, %d), length too large (max file size = %d)" %(pin_offset, pin_length, apk_file_size))
pin_length = pin_length + remaining_size
remark += '[clamped.length]'
yield InputRecord(pinlist_meta_parent, pin_offset, pin_length, remark)
def write_text_file_output(input_records: Iterable[InputRecord], output_text_file):
for rec in input_records:
output_text_file.write("%s %d %d #%s\n" %(rec.filepath, rec.offset, rec.length, rec.remark))
def build_trace_file(input_records: Iterable[InputRecord]) -> TraceFile:
trace_file = TraceFile()
trace_file_index = trace_file.index
file_id_counter = 0
file_id_map = {} # filename -> id
stats_length_total = 0
filename_stats = {} # filename -> total size
for rec in input_records:
filename = rec.filepath
file_id = file_id_map.get(filename)
if not file_id:
file_id = file_id_counter
file_id_map[filename] = file_id_counter
file_id_counter = file_id_counter + 1
file_index_entry = trace_file_index.entries.add() = file_id
file_index_entry.file_name = filename
# already in the file index, add the file entry.
file_entry = trace_file.list.entries.add()
file_entry.index_id = file_id
file_entry.file_length = rec.length
stats_length_total += file_entry.file_length
file_entry.file_offset = rec.offset
filename_stats[filename] = filename_stats.get(filename, 0) + file_entry.file_length
return trace_file
def main():
global _debug
options= parse_options()
_debug = options.debug
_debug_print("parsed options: ", options)
if not options.input:
input_file = sys.stdin
_debug_print("input = stdin")
input_file = open(options.input)
_debug_print("input = (file)", options.input)
if not options.output_proto_binary:
output_proto_file = None
output_proto_file = open(options.output_proto_binary, 'wb')
_debug_print("output_proto_binary = ", output_proto_file)
pinlist_meta_parent = options.pinlist_meta_parent
if options.pinlist_meta:
pin_meta_file = open(options.pinlist_meta, 'rb')
pin_meta_file = None
if (pinlist_meta_parent == None) != (pin_meta_file == None):
print("Options must be used together: --pinlist-meta and --pinlist-meta-path")
return 1
if not options.output_text:
output_text_file = sys.stdout
_debug_print("output = stdout")
output_text_file = open(options.output_text, 'w')
_debug_print("output = (file)", options.output_text)
zip_paths = options.zip_path or []
input_records = list(parse_pin_meta(pin_meta_file, pinlist_meta_parent, zip_paths))
input_records = input_records + list(parse_input_file(input_file, zip_paths))
for p in input_records:
write_text_file_output(input_records, output_text_file)
out_proto = build_trace_file(input_records)
if output_proto_file:
return 0
if __name__ == '__main__':