#!/usr/bin/python # # Copyright (c) 2013 The Chromium OS Authors. All rights reserved. # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. """Unit testing checker.py.""" import array import collections import cStringIO import hashlib import itertools import os import unittest # Pylint cannot find mox. # pylint: disable=F0401 import mox import checker import common import payload as update_payload # avoid name conflicts later. import test_utils import update_metadata_pb2 _PRIVKEY_FILE_NAME = 'payload-test-key.pem' _PUBKEY_FILE_NAME = 'payload-test-key.pub' def _OpTypeByName(op_name): op_name_to_type = { 'REPLACE': common.OpType.REPLACE, 'REPLACE_BZ': common.OpType.REPLACE_BZ, 'MOVE': common.OpType.MOVE, 'BSDIFF': common.OpType.BSDIFF, } return op_name_to_type[op_name] def _KiB(count): """Return the byte size of a given number of (binary) kilobytes.""" return count << 10 def _MiB(count): """Return the byte size of a given number of (binary) megabytes.""" return count << 20 def _GiB(count): """Return the byte size of a given number of (binary) gigabytes.""" return count << 30 def _GetPayloadChecker(payload_gen_write_to_file_func, *largs, **dargs): """Returns a payload checker from a given payload generator.""" payload_file = cStringIO.StringIO() payload_gen_write_to_file_func(payload_file, *largs, **dargs) payload_file.seek(0) payload = update_payload.Payload(payload_file) payload.Init() return checker.PayloadChecker(payload) def _GetPayloadCheckerWithData(payload_gen): """Returns a payload checker from a given payload generator.""" payload_file = cStringIO.StringIO() payload_gen.WriteToFile(payload_file) payload_file.seek(0) payload = update_payload.Payload(payload_file) payload.Init() return checker.PayloadChecker(payload) # (i) this class doesn't need an __init__(); (ii) unit testing is all about # running protected methods; (iii) don't bark about missing members of classes # you cannot import. # pylint: disable=W0232 # pylint: disable=W0212 # pylint: disable=E1101 class PayloadCheckerTest(mox.MoxTestBase): """Tests the PayloadChecker class. In addition to ordinary testFoo() methods, which are automatically invoked by the unittest framework, in this class we make use of DoBarTest() calls that implement parametric tests of certain features. In order to invoke each test, which embodies a unique combination of parameter values, as a complete unit test, we perform explicit enumeration of the parameter space and create individual invocation contexts for each, which are then bound as testBar__param1=val1__param2=val2(). The enumeration of parameter spaces for all such tests is done in AddAllParametricTests(). """ def MockPayload(self): """Create a mock payload object, complete with a mock menifest.""" payload = self.mox.CreateMock(update_payload.Payload) payload.is_init = True payload.manifest = self.mox.CreateMock( update_metadata_pb2.DeltaArchiveManifest) return payload @staticmethod def NewExtent(start_block, num_blocks): """Returns an Extent message. Each of the provided fields is set iff it is >= 0; otherwise, it's left at its default state. Args: start_block: the starting block of the extent num_blocks: the number of blocks in the extent Returns: An Extent message. """ ex = update_metadata_pb2.Extent() if start_block >= 0: ex.start_block = start_block if num_blocks >= 0: ex.num_blocks = num_blocks return ex @staticmethod def NewExtentList(*args): """Returns an list of extents. Args: *args: (start_block, num_blocks) pairs defining the extents Returns: A list of Extent objects. """ ex_list = [] for start_block, num_blocks in args: ex_list.append(PayloadCheckerTest.NewExtent(start_block, num_blocks)) return ex_list @staticmethod def AddToMessage(repeated_field, field_vals): for field_val in field_vals: new_field = repeated_field.add() new_field.CopyFrom(field_val) def assertIsNone(self, val): """Asserts that val is None (TODO remove once we upgrade to Python 2.7). Note that we're using assertEqual so as for it to show us the actual non-None value. Args: val: value/object to be equated to None """ self.assertEqual(val, None) def SetupAddElemTest(self, is_present, is_submsg, convert=str, linebreak=False, indent=0): """Setup for testing of _CheckElem() and its derivatives. Args: is_present: whether or not the element is found in the message is_submsg: whether the element is a sub-message itself convert: a representation conversion function linebreak: whether or not a linebreak is to be used in the report indent: indentation used for the report Returns: msg: a mock message object report: a mock report object subreport: a mock sub-report object name: an element name to check val: expected element value """ name = 'foo' val = 'fake submsg' if is_submsg else 'fake field' subreport = 'fake subreport' # Create a mock message. msg = self.mox.CreateMock(update_metadata_pb2.message.Message) msg.HasField(name).AndReturn(is_present) setattr(msg, name, val) # Create a mock report. report = self.mox.CreateMock(checker._PayloadReport) if is_present: if is_submsg: report.AddSubReport(name).AndReturn(subreport) else: report.AddField(name, convert(val), linebreak=linebreak, indent=indent) self.mox.ReplayAll() return (msg, report, subreport, name, val) def DoAddElemTest(self, is_present, is_mandatory, is_submsg, convert, linebreak, indent): """Parametric testing of _CheckElem(). Args: is_present: whether or not the element is found in the message is_mandatory: whether or not it's a mandatory element is_submsg: whether the element is a sub-message itself convert: a representation conversion function linebreak: whether or not a linebreak is to be used in the report indent: indentation used for the report """ msg, report, subreport, name, val = self.SetupAddElemTest( is_present, is_submsg, convert, linebreak, indent) largs = [msg, name, report, is_mandatory, is_submsg] dargs = {'convert': convert, 'linebreak': linebreak, 'indent': indent} if is_mandatory and not is_present: self.assertRaises(update_payload.PayloadError, checker.PayloadChecker._CheckElem, *largs, **dargs) else: ret_val, ret_subreport = checker.PayloadChecker._CheckElem(*largs, **dargs) self.assertEquals(ret_val, val if is_present else None) self.assertEquals(ret_subreport, subreport if is_present and is_submsg else None) def DoAddFieldTest(self, is_mandatory, is_present, convert, linebreak, indent): """Parametric testing of _Check{Mandatory,Optional}Field(). Args: is_mandatory: whether we're testing a mandatory call is_present: whether or not the element is found in the message convert: a representation conversion function linebreak: whether or not a linebreak is to be used in the report indent: indentation used for the report """ msg, report, _, name, val = self.SetupAddElemTest( is_present, False, convert, linebreak, indent) # Prepare for invocation of the tested method. largs = [msg, name, report] dargs = {'convert': convert, 'linebreak': linebreak, 'indent': indent} if is_mandatory: largs.append('bar') tested_func = checker.PayloadChecker._CheckMandatoryField else: tested_func = checker.PayloadChecker._CheckOptionalField # Test the method call. if is_mandatory and not is_present: self.assertRaises(update_payload.PayloadError, tested_func, *largs, **dargs) else: ret_val = tested_func(*largs, **dargs) self.assertEquals(ret_val, val if is_present else None) def DoAddSubMsgTest(self, is_mandatory, is_present): """Parametrized testing of _Check{Mandatory,Optional}SubMsg(). Args: is_mandatory: whether we're testing a mandatory call is_present: whether or not the element is found in the message """ msg, report, subreport, name, val = self.SetupAddElemTest(is_present, True) # Prepare for invocation of the tested method. largs = [msg, name, report] if is_mandatory: largs.append('bar') tested_func = checker.PayloadChecker._CheckMandatorySubMsg else: tested_func = checker.PayloadChecker._CheckOptionalSubMsg # Test the method call. if is_mandatory and not is_present: self.assertRaises(update_payload.PayloadError, tested_func, *largs) else: ret_val, ret_subreport = tested_func(*largs) self.assertEquals(ret_val, val if is_present else None) self.assertEquals(ret_subreport, subreport if is_present else None) def testCheckPresentIff(self): """Tests _CheckPresentIff().""" self.assertIsNone(checker.PayloadChecker._CheckPresentIff( None, None, 'foo', 'bar', 'baz')) self.assertIsNone(checker.PayloadChecker._CheckPresentIff( 'a', 'b', 'foo', 'bar', 'baz')) self.assertRaises(update_payload.PayloadError, checker.PayloadChecker._CheckPresentIff, 'a', None, 'foo', 'bar', 'baz') self.assertRaises(update_payload.PayloadError, checker.PayloadChecker._CheckPresentIff, None, 'b', 'foo', 'bar', 'baz') def DoCheckSha256SignatureTest(self, expect_pass, expect_subprocess_call, sig_data, sig_asn1_header, returned_signed_hash, expected_signed_hash): """Parametric testing of _CheckSha256SignatureTest(). Args: expect_pass: whether or not it should pass expect_subprocess_call: whether to expect the openssl call to happen sig_data: the signature raw data sig_asn1_header: the ASN1 header returned_signed_hash: the signed hash data retuned by openssl expected_signed_hash: the signed hash data to compare against """ # Stub out the subprocess invocation. self.mox.StubOutWithMock(checker.PayloadChecker, '_Run') if expect_subprocess_call: checker.PayloadChecker._Run(mox.IsA(list), send_data=sig_data).AndReturn( (sig_asn1_header + returned_signed_hash, None)) self.mox.ReplayAll() if expect_pass: self.assertIsNone(checker.PayloadChecker._CheckSha256Signature( sig_data, 'foo', expected_signed_hash, 'bar')) else: self.assertRaises(update_payload.PayloadError, checker.PayloadChecker._CheckSha256Signature, sig_data, 'foo', expected_signed_hash, 'bar') self.mox.UnsetStubs() def testCheckSha256Signature_Pass(self): """Tests _CheckSha256Signature(); pass case.""" sig_data = 'fake-signature'.ljust(256) signed_hash = hashlib.sha256('fake-data').digest() self.DoCheckSha256SignatureTest(True, True, sig_data, common.SIG_ASN1_HEADER, signed_hash, signed_hash) def testCheckSha256Signature_FailBadSignature(self): """Tests _CheckSha256Signature(); fails due to malformed signature.""" sig_data = 'fake-signature' # malformed (not 256 bytes in length) signed_hash = hashlib.sha256('fake-data').digest() self.DoCheckSha256SignatureTest(False, False, sig_data, common.SIG_ASN1_HEADER, signed_hash, signed_hash) def testCheckSha256Signature_FailBadOutputLength(self): """Tests _CheckSha256Signature(); fails due to unexpected output length.""" sig_data = 'fake-signature'.ljust(256) signed_hash = 'fake-hash' # malformed (not 32 bytes in length) self.DoCheckSha256SignatureTest(False, True, sig_data, common.SIG_ASN1_HEADER, signed_hash, signed_hash) def testCheckSha256Signature_FailBadAsnHeader(self): """Tests _CheckSha256Signature(); fails due to bad ASN1 header.""" sig_data = 'fake-signature'.ljust(256) signed_hash = hashlib.sha256('fake-data').digest() bad_asn1_header = 'bad-asn-header'.ljust(len(common.SIG_ASN1_HEADER)) self.DoCheckSha256SignatureTest(False, True, sig_data, bad_asn1_header, signed_hash, signed_hash) def testCheckSha256Signature_FailBadHash(self): """Tests _CheckSha256Signature(); fails due to bad hash returned.""" sig_data = 'fake-signature'.ljust(256) expected_signed_hash = hashlib.sha256('fake-data').digest() returned_signed_hash = hashlib.sha256('bad-fake-data').digest() self.DoCheckSha256SignatureTest(False, True, sig_data, common.SIG_ASN1_HEADER, expected_signed_hash, returned_signed_hash) def testCheckBlocksFitLength_Pass(self): """Tests _CheckBlocksFitLength(); pass case.""" self.assertIsNone(checker.PayloadChecker._CheckBlocksFitLength( 64, 4, 16, 'foo')) self.assertIsNone(checker.PayloadChecker._CheckBlocksFitLength( 60, 4, 16, 'foo')) self.assertIsNone(checker.PayloadChecker._CheckBlocksFitLength( 49, 4, 16, 'foo')) self.assertIsNone(checker.PayloadChecker._CheckBlocksFitLength( 48, 3, 16, 'foo')) def testCheckBlocksFitLength_TooManyBlocks(self): """Tests _CheckBlocksFitLength(); fails due to excess blocks.""" self.assertRaises(update_payload.PayloadError, checker.PayloadChecker._CheckBlocksFitLength, 64, 5, 16, 'foo') self.assertRaises(update_payload.PayloadError, checker.PayloadChecker._CheckBlocksFitLength, 60, 5, 16, 'foo') self.assertRaises(update_payload.PayloadError, checker.PayloadChecker._CheckBlocksFitLength, 49, 5, 16, 'foo') self.assertRaises(update_payload.PayloadError, checker.PayloadChecker._CheckBlocksFitLength, 48, 4, 16, 'foo') def testCheckBlocksFitLength_TooFewBlocks(self): """Tests _CheckBlocksFitLength(); fails due to insufficient blocks.""" self.assertRaises(update_payload.PayloadError, checker.PayloadChecker._CheckBlocksFitLength, 64, 3, 16, 'foo') self.assertRaises(update_payload.PayloadError, checker.PayloadChecker._CheckBlocksFitLength, 60, 3, 16, 'foo') self.assertRaises(update_payload.PayloadError, checker.PayloadChecker._CheckBlocksFitLength, 49, 3, 16, 'foo') self.assertRaises(update_payload.PayloadError, checker.PayloadChecker._CheckBlocksFitLength, 48, 2, 16, 'foo') def DoCheckManifestTest(self, fail_mismatched_block_size, fail_bad_sigs, fail_mismatched_oki_ori, fail_bad_oki, fail_bad_ori, fail_bad_nki, fail_bad_nri, fail_missing_ops): """Parametric testing of _CheckManifest(). Args: fail_mismatched_block_size: simulate a missing block_size field fail_bad_sigs: make signatures descriptor inconsistent fail_mismatched_oki_ori: make old rootfs/kernel info partially present fail_bad_oki: tamper with old kernel info fail_bad_ori: tamper with old rootfs info fail_bad_nki: tamper with new kernel info fail_bad_nri: tamper with new rootfs info fail_missing_ops: simulate a manifest without any operations """ # Generate a test payload. For this test, we only care about the manifest # and don't need any data blobs, hence we can use a plain paylaod generator # (which also gives us more control on things that can be screwed up). payload_gen = test_utils.PayloadGenerator() # Tamper with block size, if required. if fail_mismatched_block_size: payload_gen.SetBlockSize(_KiB(1)) else: payload_gen.SetBlockSize(_KiB(4)) # Add some operations. if not fail_missing_ops: payload_gen.AddOperation(False, common.OpType.MOVE, src_extents=[(0, 16), (16, 497)], dst_extents=[(16, 496), (0, 16)]) payload_gen.AddOperation(True, common.OpType.MOVE, src_extents=[(0, 8), (8, 8)], dst_extents=[(8, 8), (0, 8)]) # Set an invalid signatures block (offset but no size), if required. if fail_bad_sigs: payload_gen.SetSignatures(32, None) # Add old kernel/rootfs partition info, as required. if fail_mismatched_oki_ori or fail_bad_oki: oki_hash = (None if fail_bad_oki else hashlib.sha256('fake-oki-content').digest()) payload_gen.SetPartInfo(True, False, _KiB(512), oki_hash) if not fail_mismatched_oki_ori and fail_bad_ori: payload_gen.SetPartInfo(False, False, _MiB(8), None) # Add new kernel/rootfs partition info. payload_gen.SetPartInfo( True, True, _KiB(512), None if fail_bad_nki else hashlib.sha256('fake-nki-content').digest()) payload_gen.SetPartInfo( False, True, _MiB(8), None if fail_bad_nri else hashlib.sha256('fake-nri-content').digest()) # Create the test object. payload_checker = _GetPayloadChecker(payload_gen.WriteToFile) report = checker._PayloadReport() should_fail = (fail_mismatched_block_size or fail_bad_sigs or fail_mismatched_oki_ori or fail_bad_oki or fail_bad_ori or fail_bad_nki or fail_bad_nri or fail_missing_ops) if should_fail: self.assertRaises(update_payload.PayloadError, payload_checker._CheckManifest, report) else: self.assertIsNone(payload_checker._CheckManifest(report)) def testCheckLength(self): """Tests _CheckLength().""" payload_checker = checker.PayloadChecker(self.MockPayload()) block_size = payload_checker.block_size # Passes. self.assertIsNone(payload_checker._CheckLength( int(3.5 * block_size), 4, 'foo', 'bar')) # Fails, too few blocks. self.assertRaises(update_payload.PayloadError, payload_checker._CheckLength, int(3.5 * block_size), 3, 'foo', 'bar') # Fails, too many blocks. self.assertRaises(update_payload.PayloadError, payload_checker._CheckLength, int(3.5 * block_size), 5, 'foo', 'bar') def testCheckExtents(self): """Tests _CheckExtents().""" payload_checker = checker.PayloadChecker(self.MockPayload()) block_size = payload_checker.block_size # Passes w/ all real extents. extents = self.NewExtentList((0, 4), (8, 3), (1024, 16)) self.assertEquals( payload_checker._CheckExtents(extents, (1024 + 16) * block_size, collections.defaultdict(int), 'foo'), 23) # Passes w/ pseudo-extents (aka sparse holes). extents = self.NewExtentList((0, 4), (common.PSEUDO_EXTENT_MARKER, 5), (8, 3)) self.assertEquals( payload_checker._CheckExtents(extents, (1024 + 16) * block_size, collections.defaultdict(int), 'foo', allow_pseudo=True), 12) # Passes w/ pseudo-extent due to a signature. extents = self.NewExtentList((common.PSEUDO_EXTENT_MARKER, 2)) self.assertEquals( payload_checker._CheckExtents(extents, (1024 + 16) * block_size, collections.defaultdict(int), 'foo', allow_signature=True), 2) # Fails, extent missing a start block. extents = self.NewExtentList((-1, 4), (8, 3), (1024, 16)) self.assertRaises( update_payload.PayloadError, payload_checker._CheckExtents, extents, (1024 + 16) * block_size, collections.defaultdict(int), 'foo') # Fails, extent missing block count. extents = self.NewExtentList((0, -1), (8, 3), (1024, 16)) self.assertRaises( update_payload.PayloadError, payload_checker._CheckExtents, extents, (1024 + 16) * block_size, collections.defaultdict(int), 'foo') # Fails, extent has zero blocks. extents = self.NewExtentList((0, 4), (8, 3), (1024, 0)) self.assertRaises( update_payload.PayloadError, payload_checker._CheckExtents, extents, (1024 + 16) * block_size, collections.defaultdict(int), 'foo') # Fails, extent exceeds partition boundaries. extents = self.NewExtentList((0, 4), (8, 3), (1024, 16)) self.assertRaises( update_payload.PayloadError, payload_checker._CheckExtents, extents, (1024 + 15) * block_size, collections.defaultdict(int), 'foo') def testCheckReplaceOperation(self): """Tests _CheckReplaceOperation() where op.type == REPLACE.""" payload_checker = checker.PayloadChecker(self.MockPayload()) block_size = payload_checker.block_size data_length = 10000 op = self.mox.CreateMock( update_metadata_pb2.DeltaArchiveManifest.InstallOperation) op.type = common.OpType.REPLACE # Pass. op.src_extents = [] self.assertIsNone( payload_checker._CheckReplaceOperation( op, data_length, (data_length + block_size - 1) / block_size, 'foo')) # Fail, src extents founds. op.src_extents = ['bar'] self.assertRaises( update_payload.PayloadError, payload_checker._CheckReplaceOperation, op, data_length, (data_length + block_size - 1) / block_size, 'foo') # Fail, missing data. op.src_extents = [] self.assertRaises( update_payload.PayloadError, payload_checker._CheckReplaceOperation, op, None, (data_length + block_size - 1) / block_size, 'foo') # Fail, length / block number mismatch. op.src_extents = ['bar'] self.assertRaises( update_payload.PayloadError, payload_checker._CheckReplaceOperation, op, data_length, (data_length + block_size - 1) / block_size + 1, 'foo') def testCheckReplaceBzOperation(self): """Tests _CheckReplaceOperation() where op.type == REPLACE_BZ.""" payload_checker = checker.PayloadChecker(self.MockPayload()) block_size = payload_checker.block_size data_length = block_size * 3 op = self.mox.CreateMock( update_metadata_pb2.DeltaArchiveManifest.InstallOperation) op.type = common.OpType.REPLACE_BZ # Pass. op.src_extents = [] self.assertIsNone( payload_checker._CheckReplaceOperation( op, data_length, (data_length + block_size - 1) / block_size + 5, 'foo')) # Fail, src extents founds. op.src_extents = ['bar'] self.assertRaises( update_payload.PayloadError, payload_checker._CheckReplaceOperation, op, data_length, (data_length + block_size - 1) / block_size + 5, 'foo') # Fail, missing data. op.src_extents = [] self.assertRaises( update_payload.PayloadError, payload_checker._CheckReplaceOperation, op, None, (data_length + block_size - 1) / block_size, 'foo') # Fail, too few blocks to justify BZ. op.src_extents = [] self.assertRaises( update_payload.PayloadError, payload_checker._CheckReplaceOperation, op, data_length, (data_length + block_size - 1) / block_size, 'foo') def testCheckMoveOperation_Pass(self): """Tests _CheckMoveOperation(); pass case.""" payload_checker = checker.PayloadChecker(self.MockPayload()) op = update_metadata_pb2.DeltaArchiveManifest.InstallOperation() op.type = common.OpType.MOVE self.AddToMessage(op.src_extents, self.NewExtentList((0, 4), (12, 2), (1024, 128))) self.AddToMessage(op.dst_extents, self.NewExtentList((16, 128), (512, 6))) self.assertIsNone( payload_checker._CheckMoveOperation(op, None, 134, 134, 'foo')) def testCheckMoveOperation_FailContainsData(self): """Tests _CheckMoveOperation(); fails, message contains data.""" payload_checker = checker.PayloadChecker(self.MockPayload()) op = update_metadata_pb2.DeltaArchiveManifest.InstallOperation() op.type = common.OpType.MOVE self.AddToMessage(op.src_extents, self.NewExtentList((0, 4), (12, 2), (1024, 128))) self.AddToMessage(op.dst_extents, self.NewExtentList((16, 128), (512, 6))) self.assertRaises( update_payload.PayloadError, payload_checker._CheckMoveOperation, op, 1024, 134, 134, 'foo') def testCheckMoveOperation_FailInsufficientSrcBlocks(self): """Tests _CheckMoveOperation(); fails, not enough actual src blocks.""" payload_checker = checker.PayloadChecker(self.MockPayload()) op = update_metadata_pb2.DeltaArchiveManifest.InstallOperation() op.type = common.OpType.MOVE self.AddToMessage(op.src_extents, self.NewExtentList((0, 4), (12, 2), (1024, 127))) self.AddToMessage(op.dst_extents, self.NewExtentList((16, 128), (512, 6))) self.assertRaises( update_payload.PayloadError, payload_checker._CheckMoveOperation, op, None, 134, 134, 'foo') def testCheckMoveOperation_FailInsufficientDstBlocks(self): """Tests _CheckMoveOperation(); fails, not enough actual dst blocks.""" payload_checker = checker.PayloadChecker(self.MockPayload()) op = update_metadata_pb2.DeltaArchiveManifest.InstallOperation() op.type = common.OpType.MOVE self.AddToMessage(op.src_extents, self.NewExtentList((0, 4), (12, 2), (1024, 128))) self.AddToMessage(op.dst_extents, self.NewExtentList((16, 128), (512, 5))) self.assertRaises( update_payload.PayloadError, payload_checker._CheckMoveOperation, op, None, 134, 134, 'foo') def testCheckMoveOperation_FailExcessSrcBlocks(self): """Tests _CheckMoveOperation(); fails, too many actual src blocks.""" payload_checker = checker.PayloadChecker(self.MockPayload()) op = update_metadata_pb2.DeltaArchiveManifest.InstallOperation() op.type = common.OpType.MOVE self.AddToMessage(op.src_extents, self.NewExtentList((0, 4), (12, 2), (1024, 128))) self.AddToMessage(op.dst_extents, self.NewExtentList((16, 128), (512, 5))) self.assertRaises( update_payload.PayloadError, payload_checker._CheckMoveOperation, op, None, 134, 134, 'foo') self.AddToMessage(op.src_extents, self.NewExtentList((0, 4), (12, 2), (1024, 129))) self.AddToMessage(op.dst_extents, self.NewExtentList((16, 128), (512, 6))) self.assertRaises( update_payload.PayloadError, payload_checker._CheckMoveOperation, op, None, 134, 134, 'foo') def testCheckMoveOperation_FailExcessDstBlocks(self): """Tests _CheckMoveOperation(); fails, too many actual dst blocks.""" payload_checker = checker.PayloadChecker(self.MockPayload()) op = update_metadata_pb2.DeltaArchiveManifest.InstallOperation() op.type = common.OpType.MOVE self.AddToMessage(op.src_extents, self.NewExtentList((0, 4), (12, 2), (1024, 128))) self.AddToMessage(op.dst_extents, self.NewExtentList((16, 128), (512, 7))) self.assertRaises( update_payload.PayloadError, payload_checker._CheckMoveOperation, op, None, 134, 134, 'foo') def testCheckMoveOperation_FailStagnantBlocks(self): """Tests _CheckMoveOperation(); fails, there are blocks that do not move.""" payload_checker = checker.PayloadChecker(self.MockPayload()) op = update_metadata_pb2.DeltaArchiveManifest.InstallOperation() op.type = common.OpType.MOVE self.AddToMessage(op.src_extents, self.NewExtentList((0, 4), (12, 2), (1024, 128))) self.AddToMessage(op.dst_extents, self.NewExtentList((8, 128), (512, 6))) self.assertRaises( update_payload.PayloadError, payload_checker._CheckMoveOperation, op, None, 134, 134, 'foo') def testCheckBsdiff(self): """Tests _CheckMoveOperation().""" payload_checker = checker.PayloadChecker(self.MockPayload()) # Pass. self.assertIsNone( payload_checker._CheckBsdiffOperation(10000, 3, 'foo')) # Fail, missing data blob. self.assertRaises( update_payload.PayloadError, payload_checker._CheckBsdiffOperation, None, 3, 'foo') # Fail, too big of a diff blob (unjustified). self.assertRaises( update_payload.PayloadError, payload_checker._CheckBsdiffOperation, 10000, 2, 'foo') def DoCheckOperationTest(self, op_type_name, is_last, allow_signature, allow_unhashed, fail_src_extents, fail_dst_extents, fail_mismatched_data_offset_length, fail_missing_dst_extents, fail_src_length, fail_dst_length, fail_data_hash, fail_prev_data_offset): """Parametric testing of _CheckOperation(). Args: op_type_name: 'REPLACE', 'REPLACE_BZ', 'MOVE' or 'BSDIFF' is_last: whether we're testing the last operation in a sequence allow_signature: whether we're testing a signature-capable operation allow_unhashed: whether we're allowing to not hash the data fail_src_extents: tamper with src extents fail_dst_extents: tamper with dst extents fail_mismatched_data_offset_length: make data_{offset,length} inconsistent fail_missing_dst_extents: do not include dst extents fail_src_length: make src length inconsistent fail_dst_length: make dst length inconsistent fail_data_hash: tamper with the data blob hash fail_prev_data_offset: make data space uses incontiguous """ op_type = _OpTypeByName(op_type_name) # Create the test object. payload = self.MockPayload() payload_checker = checker.PayloadChecker(payload) block_size = payload_checker.block_size # Create auxiliary arguments. old_part_size = _MiB(4) new_part_size = _MiB(8) old_block_counters = array.array( 'B', [0] * ((old_part_size + block_size - 1) / block_size)) new_block_counters = array.array( 'B', [0] * ((new_part_size + block_size - 1) / block_size)) prev_data_offset = 1876 blob_hash_counts = collections.defaultdict(int) # Create the operation object for the test. op = update_metadata_pb2.DeltaArchiveManifest.InstallOperation() op.type = op_type total_src_blocks = 0 if op_type in (common.OpType.MOVE, common.OpType.BSDIFF): if fail_src_extents: self.AddToMessage(op.src_extents, self.NewExtentList((0, 0))) else: self.AddToMessage(op.src_extents, self.NewExtentList((0, 16))) total_src_blocks = 16 if op_type != common.OpType.MOVE: if not fail_mismatched_data_offset_length: op.data_length = 16 * block_size - 8 if fail_prev_data_offset: op.data_offset = prev_data_offset + 16 else: op.data_offset = prev_data_offset fake_data = 'fake-data'.ljust(op.data_length) if not (allow_unhashed or (is_last and allow_signature and op_type == common.OpType.REPLACE)): if not fail_data_hash: # Create a valid data blob hash. op.data_sha256_hash = hashlib.sha256(fake_data).digest() payload.ReadDataBlob(op.data_offset, op.data_length).AndReturn( fake_data) elif fail_data_hash: # Create an invalid data blob hash. op.data_sha256_hash = hashlib.sha256( fake_data.replace(' ', '-')).digest() payload.ReadDataBlob(op.data_offset, op.data_length).AndReturn( fake_data) total_dst_blocks = 0 if not fail_missing_dst_extents: total_dst_blocks = 16 if fail_dst_extents: self.AddToMessage(op.dst_extents, self.NewExtentList((4, 16), (32, 0))) else: self.AddToMessage(op.dst_extents, self.NewExtentList((4, 8), (64, 8))) if total_src_blocks: if fail_src_length: op.src_length = total_src_blocks * block_size + 8 else: op.src_length = total_src_blocks * block_size elif fail_src_length: # Add an orphaned src_length. op.src_length = 16 if total_dst_blocks: if fail_dst_length: op.dst_length = total_dst_blocks * block_size + 8 else: op.dst_length = total_dst_blocks * block_size self.mox.ReplayAll() should_fail = (fail_src_extents or fail_dst_extents or fail_mismatched_data_offset_length or fail_missing_dst_extents or fail_src_length or fail_dst_length or fail_data_hash or fail_prev_data_offset) largs = [op, 'foo', is_last, old_block_counters, new_block_counters, old_part_size, new_part_size, prev_data_offset, allow_signature, allow_unhashed, blob_hash_counts] if should_fail: self.assertRaises(update_payload.PayloadError, payload_checker._CheckOperation, *largs) else: self.assertEqual(payload_checker._CheckOperation(*largs), op.data_length if op.HasField('data_length') else 0) def testAllocBlockCounters(self): """Tests _CheckMoveOperation().""" payload_checker = checker.PayloadChecker(self.MockPayload()) block_size = payload_checker.block_size # Check allocation for block-aligned partition size, ensure it's integers. result = payload_checker._AllocBlockCounters(16 * block_size) self.assertEqual(len(result), 16) self.assertEqual(type(result[0]), int) # Check allocation of unaligned partition sizes. result = payload_checker._AllocBlockCounters(16 * block_size - 1) self.assertEqual(len(result), 16) result = payload_checker._AllocBlockCounters(16 * block_size + 1) self.assertEqual(len(result), 17) def DoCheckOperationsTest(self, fail_bad_type, fail_nonexhaustive_full_update): # Generate a test payload. For this test, we only care about one # (arbitrary) set of operations, so we'll only be generating kernel and # test with them. payload_gen = test_utils.PayloadGenerator() block_size = _KiB(4) payload_gen.SetBlockSize(block_size) rootfs_part_size = _MiB(8) # Fake rootfs operations in a full update, tampered with as required. rootfs_op_type = common.OpType.REPLACE if fail_bad_type: # Choose a type value that's bigger than the highest valid value. for valid_op_type in common.OpType.ALL: rootfs_op_type = max(rootfs_op_type, valid_op_type) rootfs_op_type += 1 rootfs_data_length = rootfs_part_size if fail_nonexhaustive_full_update: rootfs_data_length -= block_size payload_gen.AddOperation(False, rootfs_op_type, dst_extents=[(0, rootfs_data_length / block_size)], data_offset=0, data_length=rootfs_data_length) # Create the test object. payload_checker = _GetPayloadChecker(payload_gen.WriteToFile) payload_checker.payload_type = checker._TYPE_FULL report = checker._PayloadReport() should_fail = (fail_bad_type or fail_nonexhaustive_full_update) largs = (payload_checker.payload.manifest.install_operations, report, 'foo', 0, rootfs_part_size, 0, True, False) if should_fail: self.assertRaises(update_payload.PayloadError, payload_checker._CheckOperations, *largs) else: self.assertEqual(payload_checker._CheckOperations(*largs), rootfs_data_length) def DoCheckSignaturesTest(self, fail_empty_sigs_blob, fail_missing_pseudo_op, fail_mismatched_pseudo_op, fail_sig_missing_fields, fail_unknown_sig_version, fail_incorrect_sig): # Generate a test payload. For this test, we only care about the signature # block and how it relates to the payload hash. Therefore, we're generating # a random (otherwise useless) payload for this purpose. payload_gen = test_utils.EnhancedPayloadGenerator() block_size = _KiB(4) payload_gen.SetBlockSize(block_size) rootfs_part_size = _MiB(2) payload_gen.SetPartInfo(False, True, rootfs_part_size, hashlib.sha256('fake-new-rootfs-content').digest()) payload_gen.SetPartInfo(True, True, _KiB(16), hashlib.sha256('fake-new-kernel-content').digest()) payload_gen.AddOperationWithData( False, common.OpType.REPLACE, dst_extents=[(0, rootfs_part_size / block_size)], data_blob=os.urandom(rootfs_part_size)) do_forge_pseudo_op = (fail_missing_pseudo_op or fail_mismatched_pseudo_op) do_forge_sigs_data = (do_forge_pseudo_op or fail_empty_sigs_blob or fail_sig_missing_fields or fail_unknown_sig_version or fail_incorrect_sig) sigs_data = None if do_forge_sigs_data: sigs_gen = test_utils.SignaturesGenerator() if not fail_empty_sigs_blob: if fail_sig_missing_fields: sig_data = None else: sig_data = test_utils.SignSha256('fake-payload-content', _PRIVKEY_FILE_NAME) sigs_gen.AddSig(5 if fail_unknown_sig_version else 1, sig_data) sigs_data = sigs_gen.ToBinary() payload_gen.SetSignatures(payload_gen.curr_offset, len(sigs_data)) if do_forge_pseudo_op: assert sigs_data is not None, 'should have forged signatures blob by now' sigs_len = len(sigs_data) payload_gen.AddOperation( False, common.OpType.REPLACE, data_offset=payload_gen.curr_offset / 2, data_length=sigs_len / 2, dst_extents=[(0, (sigs_len / 2 + block_size - 1) / block_size)]) # Generate payload (complete w/ signature) and create the test object. payload_checker = _GetPayloadChecker( payload_gen.WriteToFileWithData, sigs_data=sigs_data, privkey_file_name=_PRIVKEY_FILE_NAME, do_add_pseudo_operation=(not do_forge_pseudo_op)) payload_checker.payload_type = checker._TYPE_FULL report = checker._PayloadReport() # We have to check the manifest first in order to set signature attributes. payload_checker._CheckManifest(report) should_fail = (fail_empty_sigs_blob or fail_missing_pseudo_op or fail_mismatched_pseudo_op or fail_sig_missing_fields or fail_unknown_sig_version or fail_incorrect_sig) largs = (report, _PUBKEY_FILE_NAME) if should_fail: self.assertRaises(update_payload.PayloadError, payload_checker._CheckSignatures, *largs) else: self.assertIsNone(payload_checker._CheckSignatures(*largs)) def DoRunTest(self, fail_wrong_payload_type, fail_invalid_block_size, fail_mismatched_block_size, fail_excess_data): # Generate a test payload. For this test, we generate a full update that # has samle kernel and rootfs operations. Since most testing is done with # internal PayloadChecker methods that are tested elsewhere, here we only # tamper with what's actually being manipulated and/or tested in the Run() # method itself. Note that the checker doesn't verify partition hashes, so # they're safe to fake. payload_gen = test_utils.EnhancedPayloadGenerator() block_size = _KiB(4) payload_gen.SetBlockSize(block_size) kernel_part_size = _KiB(16) rootfs_part_size = _MiB(2) payload_gen.SetPartInfo(False, True, rootfs_part_size, hashlib.sha256('fake-new-rootfs-content').digest()) payload_gen.SetPartInfo(True, True, kernel_part_size, hashlib.sha256('fake-new-kernel-content').digest()) payload_gen.AddOperationWithData( False, common.OpType.REPLACE, dst_extents=[(0, rootfs_part_size / block_size)], data_blob=os.urandom(rootfs_part_size)) payload_gen.AddOperationWithData( True, common.OpType.REPLACE, dst_extents=[(0, kernel_part_size / block_size)], data_blob=os.urandom(kernel_part_size)) # Generate payload (complete w/ signature) and create the test object. payload_checker = _GetPayloadChecker( payload_gen.WriteToFileWithData, privkey_file_name=_PRIVKEY_FILE_NAME, do_add_pseudo_operation=True, is_pseudo_in_kernel=True, padding=os.urandom(1024) if fail_excess_data else None) if fail_invalid_block_size: use_block_size = block_size + 5 # not a power of two elif fail_mismatched_block_size: use_block_size = block_size * 2 # different that payload stated else: use_block_size = block_size dargs = { 'pubkey_file_name': _PUBKEY_FILE_NAME, 'assert_type': 'delta' if fail_wrong_payload_type else 'full', 'block_size': use_block_size} should_fail = (fail_wrong_payload_type or fail_invalid_block_size or fail_mismatched_block_size or fail_excess_data) if should_fail: self.assertRaises(update_payload.PayloadError, payload_checker.Run, **dargs) else: self.assertIsNone(payload_checker.Run(**dargs)) # This implements a generic API, hence the occasional unused args. # pylint: disable=W0613 def ValidateCheckOperationTest(op_type_name, is_last, allow_signature, allow_unhashed, fail_src_extents, fail_dst_extents, fail_mismatched_data_offset_length, fail_missing_dst_extents, fail_src_length, fail_dst_length, fail_data_hash, fail_prev_data_offset): """Returns True iff the combination of arguments represents a valid test.""" op_type = _OpTypeByName(op_type_name) # REPLACE/REPLACE_BZ operations don't read data from src partition. if (op_type in (common.OpType.REPLACE, common.OpType.REPLACE_BZ) and ( fail_src_extents or fail_src_length)): return False # MOVE operations don't carry data. if (op_type == common.OpType.MOVE and ( fail_mismatched_data_offset_length or fail_data_hash or fail_prev_data_offset)): return False return True def TestMethodBody(run_method_name, run_dargs): """Returns a function that invokes a named method with named arguments.""" return lambda self: getattr(self, run_method_name)(**run_dargs) def AddParametricTests(tested_method_name, arg_space, validate_func=None): """Enumerates and adds specific parametric tests to PayloadCheckerTest. This function enumerates a space of test parameters (defined by arg_space), then binds a new, unique method name in PayloadCheckerTest to a test function that gets handed the said parameters. This is a preferable approach to doing the enumeration and invocation during the tests because this way each test is treated as a complete run by the unittest framework, and so benefits from the usual setUp/tearDown mechanics. Args: tested_method_name: name of the tested PayloadChecker method arg_space: a dictionary containing variables (keys) and lists of values (values) associated with them validate_func: a function used for validating test argument combinations """ for value_tuple in itertools.product(*arg_space.itervalues()): run_dargs = dict(zip(arg_space.iterkeys(), value_tuple)) if validate_func and not validate_func(**run_dargs): continue run_method_name = 'Do%sTest' % tested_method_name test_method_name = 'test%s' % tested_method_name for arg_key, arg_val in run_dargs.iteritems(): if arg_val or type(arg_val) is int: test_method_name += '__%s=%s' % (arg_key, arg_val) setattr(PayloadCheckerTest, test_method_name, TestMethodBody(run_method_name, run_dargs)) def AddAllParametricTests(): """Enumerates and adds all parametric tests to PayloadCheckerTest.""" # Add all _CheckElem() test cases. AddParametricTests('AddElem', {'linebreak': (True, False), 'indent': (0, 1, 2), 'convert': (str, lambda s: s[::-1]), 'is_present': (True, False), 'is_mandatory': (True, False), 'is_submsg': (True, False)}) # Add all _Add{Mandatory,Optional}Field tests. AddParametricTests('AddField', {'is_mandatory': (True, False), 'linebreak': (True, False), 'indent': (0, 1, 2), 'convert': (str, lambda s: s[::-1]), 'is_present': (True, False)}) # Add all _Add{Mandatory,Optional}SubMsg tests. AddParametricTests('AddSubMsg', {'is_mandatory': (True, False), 'is_present': (True, False)}) # Add all _CheckManifest() test cases. AddParametricTests('CheckManifest', {'fail_mismatched_block_size': (True, False), 'fail_bad_sigs': (True, False), 'fail_mismatched_oki_ori': (True, False), 'fail_bad_oki': (True, False), 'fail_bad_ori': (True, False), 'fail_bad_nki': (True, False), 'fail_bad_nri': (True, False), 'fail_missing_ops': (True, False)}) # Add all _CheckOperation() test cases. AddParametricTests('CheckOperation', {'op_type_name': ('REPLACE', 'REPLACE_BZ', 'MOVE', 'BSDIFF'), 'is_last': (True, False), 'allow_signature': (True, False), 'allow_unhashed': (True, False), 'fail_src_extents': (True, False), 'fail_dst_extents': (True, False), 'fail_mismatched_data_offset_length': (True, False), 'fail_missing_dst_extents': (True, False), 'fail_src_length': (True, False), 'fail_dst_length': (True, False), 'fail_data_hash': (True, False), 'fail_prev_data_offset': (True, False)}, validate_func=ValidateCheckOperationTest) # Add all _CheckOperations() test cases. AddParametricTests('CheckOperations', {'fail_bad_type': (True, False), 'fail_nonexhaustive_full_update': (True, False)}) # Add all _CheckOperations() test cases. AddParametricTests('CheckSignatures', {'fail_empty_sigs_blob': (True, False), 'fail_missing_pseudo_op': (True, False), 'fail_mismatched_pseudo_op': (True, False), 'fail_sig_missing_fields': (True, False), 'fail_unknown_sig_version': (True, False), 'fail_incorrect_sig': (True, False)}) # Add all Run() test cases. AddParametricTests('Run', {'fail_wrong_payload_type': (True, False), 'fail_invalid_block_size': (True, False), 'fail_mismatched_block_size': (True, False), 'fail_excess_data': (True, False)}) if __name__ == '__main__': AddAllParametricTests() unittest.main()