scripts/payload_info.py - SHIFTPHONES/android_system_update_engine - Gitiles

 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 #
 # Copyright (C) 2015 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.
 #

 """payload_info: Show information about an update payload."""

 from __future__ import absolute_import
 from __future__ import print_function

 import argparse
 import sys
 import textwrap

 from six.moves import range
 import update_payload


 MAJOR_PAYLOAD_VERSION_BRILLO = 2

 def DisplayValue(key, value):
   """Print out a key, value pair with values left-aligned."""
   if value != None:
     print('%-*s %s' % (28, key + ':', value))
   else:
     raise ValueError('Cannot display an empty value.')


 def DisplayHexData(data, indent=0):
   """Print out binary data as a hex values."""
   for off in range(0, len(data), 16):
     chunk = bytearray(data[off:off + 16])
     print(' ' * indent +
           ' '.join('%.2x' % c for c in chunk) +
           '   ' * (16 - len(chunk)) +
           ' | ' +
           ''.join(chr(c) if 32 <= c < 127 else '.' for c in chunk))


 class PayloadCommand(object):
   """Show basic information about an update payload.

   This command parses an update payload and displays information from
   its header and manifest.
   """

   def __init__(self, options):
     self.options = options
     self.payload = None

   def _DisplayHeader(self):
     """Show information from the payload header."""
     header = self.payload.header
     DisplayValue('Payload version', header.version)
     DisplayValue('Manifest length', header.manifest_len)

   def _DisplayManifest(self):
     """Show information from the payload manifest."""
     manifest = self.payload.manifest
     DisplayValue('Number of partitions', len(manifest.partitions))
     for partition in manifest.partitions:
       DisplayValue('  Number of "%s" ops' % partition.partition_name,
                    len(partition.operations))
     for partition in manifest.partitions:
       DisplayValue("  Timestamp for " +
                    partition.partition_name, partition.version)
     for partition in manifest.partitions:
       DisplayValue("  COW Size for " +
                    partition.partition_name, partition.estimate_cow_size)
     DisplayValue('Block size', manifest.block_size)
     DisplayValue('Minor version', manifest.minor_version)

   def _DisplaySignatures(self):
     """Show information about the signatures from the manifest."""
     header = self.payload.header
     if header.metadata_signature_len:
       offset = header.size + header.manifest_len
       DisplayValue('Metadata signatures blob',
                    'file_offset=%d (%d bytes)' %
                    (offset, header.metadata_signature_len))
       # pylint: disable=invalid-unary-operand-type
       signatures_blob = self.payload.ReadDataBlob(
           -header.metadata_signature_len,
           header.metadata_signature_len)
       self._DisplaySignaturesBlob('Metadata', signatures_blob)
     else:
       print('No metadata signatures stored in the payload')

     manifest = self.payload.manifest
     if manifest.HasField('signatures_offset'):
       signature_msg = 'blob_offset=%d' % manifest.signatures_offset
       if manifest.signatures_size:
         signature_msg += ' (%d bytes)' % manifest.signatures_size
       DisplayValue('Payload signatures blob', signature_msg)
       signatures_blob = self.payload.ReadDataBlob(manifest.signatures_offset,
                                                   manifest.signatures_size)
       self._DisplaySignaturesBlob('Payload', signatures_blob)
     else:
       print('No payload signatures stored in the payload')

   @staticmethod
   def _DisplaySignaturesBlob(signature_name, signatures_blob):
     """Show information about the signatures blob."""
     signatures = update_payload.update_metadata_pb2.Signatures()
     signatures.ParseFromString(signatures_blob)
     print('%s signatures: (%d entries)' %
           (signature_name, len(signatures.signatures)))
     for signature in signatures.signatures:
       print('  version=%s, hex_data: (%d bytes)' %
             (signature.version if signature.HasField('version') else None,
              len(signature.data)))
       DisplayHexData(signature.data, indent=4)


   def _DisplayOps(self, name, operations):
     """Show information about the install operations from the manifest.

     The list shown includes operation type, data offset, data length, source
     extents, source length, destination extents, and destinations length.

     Args:
       name: The name you want displayed above the operation table.
       operations: The operations object that you want to display information
                   about.
     """
     def _DisplayExtents(extents, name):
       """Show information about extents."""
       num_blocks = sum([ext.num_blocks for ext in extents])
       ext_str = ' '.join(
           '(%s,%s)' % (ext.start_block, ext.num_blocks) for ext in extents)
       # Make extent list wrap around at 80 chars.
       ext_str = '\n      '.join(textwrap.wrap(ext_str, 74))
       extent_plural = 's' if len(extents) > 1 else ''
       block_plural = 's' if num_blocks > 1 else ''
       print('    %s: %d extent%s (%d block%s)' %
             (name, len(extents), extent_plural, num_blocks, block_plural))
       print('      %s' % ext_str)

     op_dict = update_payload.common.OpType.NAMES
     print('%s:' % name)
     for op_count, op in enumerate(operations):
       print('  %d: %s' % (op_count, op_dict[op.type]))
       if op.HasField('data_offset'):
         print('    Data offset: %s' % op.data_offset)
       if op.HasField('data_length'):
         print('    Data length: %s' % op.data_length)
       if op.src_extents:
         _DisplayExtents(op.src_extents, 'Source')
       if op.dst_extents:
         _DisplayExtents(op.dst_extents, 'Destination')

   def _GetStats(self, manifest):
     """Returns various statistics about a payload file.

     Returns a dictionary containing the number of blocks read during payload
     application, the number of blocks written, and the number of seeks done
     when writing during operation application.
     """
     read_blocks = 0
     written_blocks = 0
     num_write_seeks = 0
     for partition in manifest.partitions:
       last_ext = None
       for curr_op in partition.operations:
         read_blocks += sum([ext.num_blocks for ext in curr_op.src_extents])
         written_blocks += sum([ext.num_blocks for ext in curr_op.dst_extents])
         for curr_ext in curr_op.dst_extents:
           # See if the extent is contiguous with the last extent seen.
           if last_ext and (curr_ext.start_block !=
                            last_ext.start_block + last_ext.num_blocks):
             num_write_seeks += 1
           last_ext = curr_ext

       # Old and new partitions are read once during verification.
       read_blocks += partition.old_partition_info.size // manifest.block_size
       read_blocks += partition.new_partition_info.size // manifest.block_size

     stats = {'read_blocks': read_blocks,
              'written_blocks': written_blocks,
              'num_write_seeks': num_write_seeks}
     return stats

   def _DisplayStats(self, manifest):
     stats = self._GetStats(manifest)
     DisplayValue('Blocks read', stats['read_blocks'])
     DisplayValue('Blocks written', stats['written_blocks'])
     DisplayValue('Seeks when writing', stats['num_write_seeks'])

   def Run(self):
     """Parse the update payload and display information from it."""
     self.payload = update_payload.Payload(self.options.payload_file)
     self.payload.Init()
     self._DisplayHeader()
     self._DisplayManifest()
     if self.options.signatures:
       self._DisplaySignatures()
     if self.options.stats:
       self._DisplayStats(self.payload.manifest)
     if self.options.list_ops:
       print()
       for partition in self.payload.manifest.partitions:
         self._DisplayOps('%s install operations' % partition.partition_name,
                          partition.operations)


 def main():
   parser = argparse.ArgumentParser(
       description='Show information about an update payload.')
   parser.add_argument('payload_file', type=argparse.FileType('rb'),
                       help='The update payload file.')
   parser.add_argument('--list_ops', default=False, action='store_true',
                       help='List the install operations and their extents.')
   parser.add_argument('--stats', default=False, action='store_true',
                       help='Show information about overall input/output.')
   parser.add_argument('--signatures', default=False, action='store_true',
                       help='Show signatures stored in the payload.')
   args = parser.parse_args()

   PayloadCommand(args).Run()


 if __name__ == '__main__':
   sys.exit(main())
	#!/usr/bin/env python
	# -- coding: utf-8 --
	#
	# Copyright (C) 2015 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.
	#

	"""payload_info: Show information about an update payload."""

	from __future__ import absolute_import
	from __future__ import print_function

	import argparse
	import sys
	import textwrap

	from six.moves import range
	import update_payload


	MAJOR_PAYLOAD_VERSION_BRILLO = 2

	def DisplayValue(key, value):
	"""Print out a key, value pair with values left-aligned."""
	if value != None:
	print('%-*s %s' % (28, key + ':', value))
	else:
	raise ValueError('Cannot display an empty value.')


	def DisplayHexData(data, indent=0):
	"""Print out binary data as a hex values."""
	for off in range(0, len(data), 16):
	chunk = bytearray(data[off:off + 16])
	print(' ' * indent +
	' '.join('%.2x' % c for c in chunk) +
	' ' * (16 - len(chunk)) +
	' \| ' +
	''.join(chr(c) if 32 <= c < 127 else '.' for c in chunk))


	class PayloadCommand(object):
	"""Show basic information about an update payload.

	This command parses an update payload and displays information from
	its header and manifest.
	"""

	def __init__(self, options):
	self.options = options
	self.payload = None

	def _DisplayHeader(self):
	"""Show information from the payload header."""
	header = self.payload.header
	DisplayValue('Payload version', header.version)
	DisplayValue('Manifest length', header.manifest_len)

	def _DisplayManifest(self):
	"""Show information from the payload manifest."""
	manifest = self.payload.manifest
	DisplayValue('Number of partitions', len(manifest.partitions))
	for partition in manifest.partitions:
	DisplayValue(' Number of "%s" ops' % partition.partition_name,
	len(partition.operations))
	for partition in manifest.partitions:
	DisplayValue(" Timestamp for " +
	partition.partition_name, partition.version)
	for partition in manifest.partitions:
	DisplayValue(" COW Size for " +
	partition.partition_name, partition.estimate_cow_size)
	DisplayValue('Block size', manifest.block_size)
	DisplayValue('Minor version', manifest.minor_version)

	def _DisplaySignatures(self):
	"""Show information about the signatures from the manifest."""
	header = self.payload.header
	if header.metadata_signature_len:
	offset = header.size + header.manifest_len
	DisplayValue('Metadata signatures blob',
	'file_offset=%d (%d bytes)' %
	(offset, header.metadata_signature_len))
	# pylint: disable=invalid-unary-operand-type
	signatures_blob = self.payload.ReadDataBlob(
	-header.metadata_signature_len,
	header.metadata_signature_len)
	self._DisplaySignaturesBlob('Metadata', signatures_blob)
	else:
	print('No metadata signatures stored in the payload')

	manifest = self.payload.manifest
	if manifest.HasField('signatures_offset'):
	signature_msg = 'blob_offset=%d' % manifest.signatures_offset
	if manifest.signatures_size:
	signature_msg += ' (%d bytes)' % manifest.signatures_size
	DisplayValue('Payload signatures blob', signature_msg)
	signatures_blob = self.payload.ReadDataBlob(manifest.signatures_offset,
	manifest.signatures_size)
	self._DisplaySignaturesBlob('Payload', signatures_blob)
	else:
	print('No payload signatures stored in the payload')

	@staticmethod
	def _DisplaySignaturesBlob(signature_name, signatures_blob):
	"""Show information about the signatures blob."""
	signatures = update_payload.update_metadata_pb2.Signatures()
	signatures.ParseFromString(signatures_blob)
	print('%s signatures: (%d entries)' %
	(signature_name, len(signatures.signatures)))
	for signature in signatures.signatures:
	print(' version=%s, hex_data: (%d bytes)' %
	(signature.version if signature.HasField('version') else None,
	len(signature.data)))
	DisplayHexData(signature.data, indent=4)


	def _DisplayOps(self, name, operations):
	"""Show information about the install operations from the manifest.

	The list shown includes operation type, data offset, data length, source
	extents, source length, destination extents, and destinations length.

	Args:
	name: The name you want displayed above the operation table.
	operations: The operations object that you want to display information
	about.
	"""
	def _DisplayExtents(extents, name):
	"""Show information about extents."""
	num_blocks = sum([ext.num_blocks for ext in extents])
	ext_str = ' '.join(
	'(%s,%s)' % (ext.start_block, ext.num_blocks) for ext in extents)
	# Make extent list wrap around at 80 chars.
	ext_str = '\n '.join(textwrap.wrap(ext_str, 74))
	extent_plural = 's' if len(extents) > 1 else ''
	block_plural = 's' if num_blocks > 1 else ''
	print(' %s: %d extent%s (%d block%s)' %
	(name, len(extents), extent_plural, num_blocks, block_plural))
	print(' %s' % ext_str)

	op_dict = update_payload.common.OpType.NAMES
	print('%s:' % name)
	for op_count, op in enumerate(operations):
	print(' %d: %s' % (op_count, op_dict[op.type]))
	if op.HasField('data_offset'):
	print(' Data offset: %s' % op.data_offset)
	if op.HasField('data_length'):
	print(' Data length: %s' % op.data_length)
	if op.src_extents:
	_DisplayExtents(op.src_extents, 'Source')
	if op.dst_extents:
	_DisplayExtents(op.dst_extents, 'Destination')

	def _GetStats(self, manifest):
	"""Returns various statistics about a payload file.

	Returns a dictionary containing the number of blocks read during payload
	application, the number of blocks written, and the number of seeks done
	when writing during operation application.
	"""
	read_blocks = 0
	written_blocks = 0
	num_write_seeks = 0
	for partition in manifest.partitions:
	last_ext = None
	for curr_op in partition.operations:
	read_blocks += sum([ext.num_blocks for ext in curr_op.src_extents])
	written_blocks += sum([ext.num_blocks for ext in curr_op.dst_extents])
	for curr_ext in curr_op.dst_extents:
	# See if the extent is contiguous with the last extent seen.
	if last_ext and (curr_ext.start_block !=
	last_ext.start_block + last_ext.num_blocks):
	num_write_seeks += 1
	last_ext = curr_ext

	# Old and new partitions are read once during verification.
	read_blocks += partition.old_partition_info.size // manifest.block_size
	read_blocks += partition.new_partition_info.size // manifest.block_size

	stats = {'read_blocks': read_blocks,
	'written_blocks': written_blocks,
	'num_write_seeks': num_write_seeks}
	return stats

	def _DisplayStats(self, manifest):
	stats = self._GetStats(manifest)
	DisplayValue('Blocks read', stats['read_blocks'])
	DisplayValue('Blocks written', stats['written_blocks'])
	DisplayValue('Seeks when writing', stats['num_write_seeks'])

	def Run(self):
	"""Parse the update payload and display information from it."""
	self.payload = update_payload.Payload(self.options.payload_file)
	self.payload.Init()
	self._DisplayHeader()
	self._DisplayManifest()
	if self.options.signatures:
	self._DisplaySignatures()
	if self.options.stats:
	self._DisplayStats(self.payload.manifest)
	if self.options.list_ops:
	print()
	for partition in self.payload.manifest.partitions:
	self._DisplayOps('%s install operations' % partition.partition_name,
	partition.operations)


	def main():
	parser = argparse.ArgumentParser(
	description='Show information about an update payload.')
	parser.add_argument('payload_file', type=argparse.FileType('rb'),
	help='The update payload file.')
	parser.add_argument('--list_ops', default=False, action='store_true',
	help='List the install operations and their extents.')
	parser.add_argument('--stats', default=False, action='store_true',
	help='Show information about overall input/output.')
	parser.add_argument('--signatures', default=False, action='store_true',
	help='Show signatures stored in the payload.')
	args = parser.parse_args()

	PayloadCommand(args).Run()


	if __name__ == '__main__':
	sys.exit(main())