Python bitstring 模块，Bits() 实例源码

def _testStringsFromFile(self):
        values = [
            'It was the summer of 95 (so what!)',
            'In the backyard, shaving the old plies',
            'Feeling so strong (strong!), something went wrong (wrong!)',
            'Straight into my finger, what a stinger, it was so long',
            'I still remember that day, like the day that I said that I swear',
            '"I\'ll never hurt myself again", but it seems that I\'m deemed to be wrong',
            'To be wrong, to be wrong',
            'Gotta keep holding on...they always played a slow song.',
        ]
        filename = './kitty_strings.txt'
        with open(filename, 'wb') as f:
            f.write('\n'.join(values))
        uut = String(name=self.uut_name, value='streetlight')
        all_mutations = self.get_all_mutations(uut)
        for value in values:
            self.assertIn(Bits(bytes=value), all_mutations)
        os.remove(filename)

def testConditionAppliesFirst(self):
        field = self.get_condition_field()
        condition = self.get_applies_first_condition()
        inner_field = String(ConditionTest.inner_field_value)
        condition_container = self.cls(condition=condition, fields=[inner_field], fuzzable=True)
        # This is done to allow field name resolution
        enclosing = Container(fields=[field, condition_container])
        self.assertEqual(condition_container.render(), inner_field.render())
        while condition_container.mutate():
            self.assertEqual(condition_container.render(), inner_field.render())

        condition_container.reset()
        field.mutate()
        self.assertEqual(condition_container.render(), Bits())
        while condition_container.mutate():
            self.assertEqual(condition_container.render(), Bits())

        del enclosing

def testConditionNotAppliesFirst(self):
        field = self.get_condition_field()
        condition = self.get_not_applies_first_condition()
        inner_field = String(ConditionTest.inner_field_value)
        condition_container = self.cls(condition=condition, fields=[inner_field], fuzzable=True)
        # This is done to allow field name resolution
        enclosing = Container(fields=[field, condition_container])
        self.assertEqual(condition_container.render(), Bits())
        while condition_container.mutate():
            self.assertEqual(condition_container.render(), Bits())

        condition_container.reset()
        field.mutate()
        self.assertEqual(condition_container.render(), inner_field.render())
        while condition_container.mutate():
            self.assertEqual(condition_container.render(), inner_field.render())

        del enclosing

def render(self, ctx=None):
        '''
        Render the current value into a :class:`bitstring.Bits` object

        :rtype: :class:`bitstring.Bits`
        :return: the rendered field
        '''
        self._initialize()
        if ctx is None:
            ctx = RenderContext()
        #
        # if we are called from within render, return a dummy object...
        #
        if self in ctx:
            self._current_rendered = self._in_render_value()
        else:
            ctx.push(self)
            if self.dependency_type == Calculated.VALUE_BASED:
                self._rendered_field = self._field.render(ctx)
            self._render()
            ctx.pop()
        return self._current_rendered

def __init__(self, depends_on, func, encoder=ENC_BITS_DEFAULT, fuzzable=True, name=None):
        '''
        :param depends_on: (name of) field we depend on
        :type encoder: :class:`~kitty.model.low_level.encoder.BitsEncoder`
        :param func: function for processing of the dependant data. func(Bits)->Bits
        :param encoder: encoder for the field
        :param fuzzable: is container fuzzable
        :param name: (unique) name of the container
        '''
        try:
            res = func(empty_bits)
            kassert.is_of_types(res, Bits)
            self._func = func
        except:
            raise KittyException('func should be func(Bits)->Bits')
        super(CalculatedBits, self).__init__(depends_on=depends_on, encoder=encoder, fuzzable=fuzzable, name=name)

def render(self, ctx=None):
        '''
        Only render if condition applies

        :param ctx: rendering context in which the method was called
        :rtype: `Bits`
        :return: rendered value of the container
        '''
        if ctx is None:
            ctx = RenderContext()
        self._initialize()
        if self in ctx:
            self._current_rendered = self._in_render_value()
        else:
            ctx.push(self)
            if self._evaluate_condition(ctx):
                super(Conditional, self).render(ctx)
            else:
                self.set_current_value(empty_bits)
            ctx.pop()
        return self._current_rendered

def __init__(self, pad_length, pad_data='\x00', fields=[], fuzzable=True, name=None):
        '''
        :param pad_length: length to pad up to (in bits)
        :param pad_data: data to pad with (default: '\x00')
        :param fields: enclosed field(s) (default: [])
        :param fuzzable: is fuzzable (default: True)
        :param name: (unique) name of the template (default: None)

        :example:

            Pad a string with ' 's so it is at least 20 bytes

            ::

                Pad(fields=String('padded'), pad_data=' ', pad_length=20)
                # default result will be: 'padded              '
        '''
        super(Pad, self).__init__(fields=fields, encoder=ENC_BITS_DEFAULT, fuzzable=fuzzable, name=name)
        self._pad_length = pad_length
        self._pad_data = Bits(bytes=pad_data)

def render(self, ctx=None):
        '''
        Render only the mutated field (or the first one if not in mutation)

        :param ctx: rendering context in which the method was called
        :rtype: `Bits`
        :return: rendered value of the container
        '''
        if ctx is None:
            ctx = RenderContext()
        ctx.push(self)
        self._initialize()
        offset = self.offset if self.offset else 0
        self._fields[self._field_idx].set_offset(offset)
        rendered = self._fields[self._field_idx].render(ctx)
        self.set_current_value(rendered)
        ctx.pop()
        return self._current_rendered

def __init__(self, value, num_bits=1, fuzzable=True, name=None):
        '''
        :param value: value to mutate (str)
        :param num_bits: number of consequtive bits to flip (invert)
        :param fuzzable: is field fuzzable (default: True)
        :param name: name of the object (default: None)

        :raises: ``KittyException`` if num_bits is bigger than the value length in bits
        :raises: ``KittyException`` if num_bits is not positive
        '''
        kassert.is_of_types(value, types.StringTypes)
        if len(value) * 8 < num_bits:
            raise KittyException('len of value in bits(%d) < num_bits(%d)' % (len(value) * 8, num_bits))
        if num_bits <= 0:
            raise KittyException('num_bits(%d) <= 0' % (num_bits))
        super(BitFlip, self).__init__(value=Bits(bytes=value), encoder=ENC_BITS_DEFAULT, fuzzable=fuzzable, name=name)
        self._data_len = len(value) * 8
        self._num_bits = num_bits
        self._num_mutations = self._data_len - (num_bits - 1)

def render(self, ctx=None):
        '''
        Render the current value into a :class:`bitstring.Bits` object

        :rtype: :class:`bitstring.Bits`
        :return: the rendered field
        '''
        self._initialize()
        if ctx is None:
            ctx = RenderContext()
        #
        # if we are called from within render, return a dummy object...
        #
        if self in ctx:
            self._current_rendered = self._in_render_value()
        else:
            ctx.push(self)
            if self.dependency_type == Calculated.VALUE_BASED:
                self._rendered_field = self._field.render(ctx)
            self._render()
            ctx.pop()
        return self._current_rendered

def __init__(self, depends_on, func, encoder=ENC_BITS_DEFAULT, fuzzable=True, name=None):
        '''
        :param depends_on: (name of) field we depend on
        :type encoder: :class:`~kitty.model.low_level.encoder.BitsEncoder`
        :param func: function for processing of the dependant data. func(Bits)->Bits
        :param encoder: encoder for the field
        :param fuzzable: is container fuzzable
        :param name: (unique) name of the container
        '''
        try:
            res = func(empty_bits)
            kassert.is_of_types(res, Bits)
            self._func = func
        except:
            raise KittyException('func should be func(Bits)->Bits')
        super(CalculatedBits, self).__init__(depends_on=depends_on, encoder=encoder, fuzzable=fuzzable, name=name)

def render(self, ctx=None):
        '''
        Only render if condition applies

        :param ctx: rendering context in which the method was called
        :rtype: `Bits`
        :return: rendered value of the container
        '''
        if ctx is None:
            ctx = RenderContext()
        self._initialize()
        if self in ctx:
            self._current_rendered = self._in_render_value()
        else:
            ctx.push(self)
            if self._evaluate_condition(ctx):
                super(Conditional, self).render(ctx)
            else:
                self.set_current_value(empty_bits)
            ctx.pop()
        return self._current_rendered

def __init__(self, pad_length, pad_data='\x00', fields=[], fuzzable=True, name=None):
        '''
        :param pad_length: length to pad up to (in bits)
        :param pad_data: data to pad with (default: '\x00')
        :param fields: enclosed field(s) (default: [])
        :param fuzzable: is fuzzable (default: True)
        :param name: (unique) name of the template (default: None)

        :example:

            Pad a string with ' 's so it is at least 20 bytes

            ::

                Pad(fields=String('padded'), pad_data=' ', pad_length=20)
                # default result will be: 'padded              '
        '''
        super(Pad, self).__init__(fields=fields, encoder=ENC_BITS_DEFAULT, fuzzable=fuzzable, name=name)
        self._pad_length = pad_length
        self._pad_data = Bits(bytes=pad_data)

def __init__(self, value, num_bits=1, fuzzable=True, name=None):
        '''
        :param value: value to mutate (str)
        :param num_bits: number of consequtive bits to flip (invert)
        :param fuzzable: is field fuzzable (default: True)
        :param name: name of the object (default: None)

        :raises: ``KittyException`` if num_bits is bigger than the value length in bits
        :raises: ``KittyException`` if num_bits is not positive
        '''
        kassert.is_of_types(value, types.StringTypes)
        if len(value) * 8 < num_bits:
            raise KittyException('len of value in bits(%d) < num_bits(%d)' % (len(value) * 8, num_bits))
        if num_bits <= 0:
            raise KittyException('num_bits(%d) <= 0' % (num_bits))
        super(BitFlip, self).__init__(value=Bits(bytes=value), encoder=ENC_BITS_DEFAULT, fuzzable=fuzzable, name=name)
        self._data_len = len(value) * 8
        self._num_bits = num_bits
        self._num_mutations = self._data_len - (num_bits - 1)

def write_bytes(self, value, nbytes=None):
        import bitstring
        # TODO: strings are utf-8 from json reading
        if isinstance(value, six.text_type):
            value = value.encode('latin-1')

        value_len = len(value)

        # Ensure the string is under the required data width
        if nbytes is None:
            nbytes = value_len
        else:
            if value_len > nbytes:
                value = value[:nbytes]
            elif value_len < nbytes:
                value += b' ' * (nbytes - value_len)

        # Cannot use string format shortcut, i.e. 'bytes:{}={}' due to the
        # automatic whitespace trimming by bitstring.
        self.bit_stream += bitstring.Bits(bytes=value)
        return value

def pack_glyph(glyph, ili9341_t3_font):
    glyph_bits = BitString()
    header = glyph['header']
    glyph_bits.append(Bits(uint=header['encoding'], length=3))
    glyph_bits.append(Bits(uint=header['width'], length=ili9341_t3_font['bits_width']))
    glyph_bits.append(Bits(uint=header['height'], length=ili9341_t3_font['bits_height']))
    glyph_bits.append(Bits(int=header['xoffset'], length=ili9341_t3_font['bits_xoffset']))
    glyph_bits.append(Bits(int=header['yoffset'], length=ili9341_t3_font['bits_yoffset']))
    glyph_bits.append(Bits(uint=header['delta'], length=ili9341_t3_font['bits_delta']))
    for row in glyph['data']:
        if row['repeat'] == 0:
            glyph_bits.append(Bits(bool=False))
        else:
            glyph_bits.append(Bits(bool=True))
            glyph_bits.append(Bits(uint=row['repeat'] - 1, length=3))

        for bit in row['bits']:
            glyph_bits.append(Bits(bool=bit == 1))
    glyph_bytes = glyph_bits.tobytes()
    return glyph_bytes

def unpack(self, data):
        if not isinstance(data, _Bits):
            data = _Bits(data)
        for i in range(self.count):
            offset = i*self.size
            try:
                b = data[offset:offset + self.size]
                self._values[i] = self.logical_range.scale_to(self.physical_range, b.int if self.logical_range.minimum<0 else b.uint)
            except ArithmeticError:
                # If the value is outside of the logical range, and NULLs are allowed, then do not modify the value
                if not self.flags & ReportFlags.NULL_STATE:
                    raise

def deserialize(self, data, report_type: ReportType):
        offset = 0
        if not isinstance(data, _Bits):
            data = _Bits(data)
        for item in self.items:
            if isinstance(item, Report):
                if item.report_type is not report_type:
                    continue
                item.unpack(data[offset:offset + item.bits])
            else:
                item.deserialize(data[offset:offset + item.bits], report_type)
            offset += item.bits

def testSessionDataNotFuzzable(self):
        field = self.cls(key=self.key_exists, default_value=self.default_value)
        self.assertEqual(self.default_value_rendered, field.render())
        field.set_session_data(self.default_session_data)
        self.assertEqual(Bits(bytes=self.value_exists), field.render())
        self.assertEqual(Bits(bytes=self.value_exists), field.render())

def testSessionDataNotFuzzableAfterReset(self):
        field = self.cls(key=self.key_exists, default_value=self.default_value)
        self.assertEqual(self.default_value_rendered, field.render())
        field.set_session_data(self.default_session_data)
        self.assertEqual(Bits(bytes=self.value_exists), field.render())
        field.reset()
        self.assertEqual(self.default_value_rendered, field.render())

def testSessionDataNotFuzzableDataChangeKeyExists(self):
        field = self.cls(key=self.key_exists, default_value=self.default_value)
        self.assertEqual(self.default_value_rendered, field.render())
        field.set_session_data(self.default_session_data)
        self.assertEqual(Bits(bytes=self.value_exists), field.render())
        new_val = 'new value'
        field.set_session_data({self.key_exists: new_val})
        self.assertEqual(Bits(bytes=new_val), field.render())

def testSessionDataNotFuzzableDataChangeKeyNotExist(self):
        field = self.cls(key=self.key_exists, default_value=self.default_value)
        self.assertEqual(self.default_value_rendered, field.render())
        field.set_session_data(self.default_session_data)
        self.assertEqual(Bits(bytes=self.value_exists), field.render())
        new_val = 'new value'
        field.set_session_data({self.key_not_exist: new_val})
        self.assertEqual(Bits(bytes=self.value_exists), field.render())

def testSessionDataFuzzableDataChangeKeyExists(self):
        field = self.cls(key=self.key_exists, default_value=self.default_value, length=len(self.default_value), fuzzable=True)
        self.assertEqual(self.default_value_rendered, field.render())
        field.set_session_data(self.default_session_data)
        self.assertEqual(Bits(bytes=self.value_exists), field.render())
        new_val = 'new value'
        field.set_session_data({self.key_exists: new_val})
        self.assertEqual(Bits(bytes=new_val), field.render())

def testSessionDataFuzzableDataChangeKeyNotExist(self):
        field = self.cls(key=self.key_exists, default_value=self.default_value, length=len(self.default_value), fuzzable=True)
        self.assertEqual(self.default_value_rendered, field.render())
        field.set_session_data(self.default_session_data)
        self.assertEqual(Bits(bytes=self.value_exists), field.render())
        new_val = 'new value'
        field.set_session_data({self.key_not_exist: new_val})
        self.assertEqual(Bits(bytes=self.value_exists), field.render())

def testMutations(self):
        field = self.get_default_field()
        mutations = self._get_all_mutations(field)
        self.assertListEqual([Bits(bytes=x) for x in self.default_values], mutations)
        mutations = self._get_all_mutations(field)
        self.assertListEqual([Bits(bytes=x) for x in self.default_values], mutations)

def testExceptionWhenFuncReturnsBits(self):
        def func(s):
            return Bits('')

        with self.assertRaises(KittyException):
            self.get_default_field(func=func)

def calculate(self, field):
        value = field.render()
        digest = self.hasher(value.bytes).digest()
        return Bits(bytes=digest)

def testInvalidHashFunctionReturnsBits(self):
        def func(data):
            return Bits()

        with self.assertRaises(KittyException):
            Hash(self.depends_on_name, algorithm='boom')

def _encode_func(self, value):
        return Bits(bytes=pack(self.fmt, value))

def testZero(self):
        uut = BitFieldMultiByteEncoder()
        self.assertEqual(uut.encode(0, 10, False), Bits(bytes='\x00'))

def testReturnValueIsBits(self):
        uut = self.get_default_encoder()
        encoded = uut.encode('abc')
        self.assertIsInstance(encoded, Bits)

def testCorrectEncoding(self):
        value = Bits(bin='01010011')
        uut = self.get_default_encoder()
        self.assertEqual(uut.encode(value), self._encode_func(value))

def _encode_func(self, bits):
        remainder = len(bits) % 8
        pad_len = (8 - remainder) % 8
        return bits + Bits(pad_len)

def testPaddingNoPad(self):
        value = Bits(bytes='\x01')
        uut = self.get_default_encoder()
        self.assertEqual(uut.encode(value), value)

def testPadding1(self):
        value = Bits(bin='1111111')
        expected = Bits(bin='11111110')
        uut = self.get_default_encoder()
        self.assertEqual(uut.encode(value), expected)

def testPadding4(self):
        value = Bits(bin='1111')
        expected = Bits(bin='11110000')
        uut = self.get_default_encoder()
        self.assertEqual(uut.encode(value), expected)

def testPadding7(self):
        value = Bits(bin='1')
        expected = Bits(bin='10000000')
        uut = self.get_default_encoder()
        self.assertEqual(uut.encode(value), expected)

def testExceptionWhenBitsNotByteAligned(self):
        value = Bits(bin='1111111')
        uut = self.get_default_encoder()
        with self.assertRaises(KittyException):
            uut.encode(value)

def _test_fields(self, init_fields=[], push_fields=[]):
        all_fields = init_fields + push_fields
        container = self.get_default_container(fields=init_fields, fuzzable=True)
        for f in push_fields:
            container.push(f)
            if isinstance(f, Container):
                # default is to pop the container immediatly in the tests...
                container.pop()
        fields_num_mutations = sum(f.num_mutations() for f in all_fields)
        container_num_mutations = container.num_mutations()
        self.assertEqual(fields_num_mutations, container_num_mutations)

        field_default_values = []
        for f in all_fields:
            field_default_values.append(f.render())
        fields_mutations = []
        for i, field in enumerate(all_fields):
            prefix = sum(field_default_values[:i])
            postfix = sum(field_default_values[i + 1:])
            if prefix == 0:
                prefix = Bits()
            if postfix == 0:
                postfix = Bits()
            while field.mutate():
                fields_mutations.append(prefix + field.render() + postfix)
            field.reset()
        container_mutations = self.get_all_mutations(container)
        self.assertListEqual(fields_mutations, container_mutations)

def testlways(self):
        field = self.get_condition_field()
        condition = self.get_not_applies_always_condition()
        condition_container = self.cls(condition=condition, fields=[String(ConditionTest.inner_field_value)], fuzzable=True)
        # This is done to allow field name resolution
        enclosing = Container(fields=[field, condition_container])
        rendered = condition_container.render()
        self.assertEqual(rendered, Bits())
        while condition_container.mutate():
            rendered = condition_container.render()
            self.assertEqual(rendered, Bits())
        del enclosing

def _testValuePadded(self, field, uut, pad_length, pad_data):
        fdata = field.render()
        udata = uut.render()
        actual_pad_len = max(0, pad_length - len(fdata))
        expected_padding = Bits(bytes=pad_data * (actual_pad_len / 8 + 1))[:actual_pad_len]
        self.assertEqual(fdata, udata[:len(fdata)])
        self.assertEqual(expected_padding, udata[len(fdata):])

def testFixedWithPad(self):
        data = 'abcd'
        expected = Bits(bytes='abcd\xff\xff\xff\xff\xff\xff')
        uut = Pad(pad_length=10 * 8, fields=Static(data), pad_data='\xff')
        uut_num_mutations = uut.num_mutations()
        self.assertEqual(uut_num_mutations, 0)
        actual_num_mutations = 0
        while uut.mutate():
            actual_num_mutations += 1
        self.assertEqual(actual_num_mutations, uut_num_mutations)
        self.assertEqual(uut.render(), expected)

def testFixedWithoutPad(self):
        data = 'abcdefghijklmnop'
        expected = Bits(bytes=data)
        uut = Pad(pad_length=10 * 8, fields=Static(data), pad_data='\xff')
        uut_num_mutations = uut.num_mutations()
        self.assertEqual(uut_num_mutations, 0)
        actual_num_mutations = 0
        while uut.mutate():
            actual_num_mutations += 1
        self.assertEqual(actual_num_mutations, uut_num_mutations)
        self.assertEqual(uut.render(), expected)

def testRendersToEmptyBits(self):
        uut = PseudoTemplate('uut')
        self.assertEquals(uut.render(), Bits())

def render(self, ctx=None):
        '''
        Render the current value of the field

        :rtype: Bits
        :return: rendered value
        '''
        self._initialize()
        if not self.is_default():
            self._current_rendered = self._encode_value(self._current_value)
        return self._current_rendered

def _mutate(self):
        if self._step:
            length = self._min_length + self._step * self._current_index
        else:
            length = self._random.randint(self._min_length, self._max_length)
        current_bytes = ''
        for i in range(length / 8 + 1):
            current_bytes += chr(self._random.randint(0, 255))
        self._current_value = Bits(bytes=current_bytes)[:length]

def _in_render_value(self):
        '''
        This method is called when rendered was called recursively.
        This means that we are enclosed by the field we depend on.
        So consider carefully what value are you going to return....

        :rtype: Bits
        :return: a dummy rendered value
        '''
        raise NotImplementedError('_in_render_value should be implemented in subclass (%s)' % type(self).__name__)

def _in_render_value(self):
        '''
        :return: a zeroed version of the field, good for some checksums and inclusive lengths
        '''
        return Bits(len(self._bit_field.render()))

def __init__(self, depends_on, length, algorithm='crc32', encoder=ENC_INT_DEFAULT, fuzzable=False, name=None):
        '''
        :param depends_on: (name of) field to be checksummed
        :param length: length of the checksum field (in bits)
        :param algorithm: checksum algorithm name (from Checksum._algos) or a function to calculate the value of the field. func(Bits) -> int
        :type encoder: :class:`~kitty.model.low_level.encoder.BitFieldEncoder`
        :param encoder: encoder for the field (default: ENC_INT_DEFAULT)
        :param fuzzable: is field fuzzable (default: False)
        :param name: (unique) name of the field (default: None)

        :example:

            ::

                Container(name='checksummed chunk', fields=[
                    RandomBytes(name='chunk', value='1234', min_length=0, max_length=75),
                    Checksum(name='CRC', depends_on='chunk', length=32)
                ])
        '''
        if algorithm in Checksum._algos:
            func = Checksum._algos[algorithm]
        else:
            try:
                res = algorithm(empty_bits)
                kassert.is_of_types(res, types.IntType)
                func = algorithm
            except:
                raise KittyException('algorithm should be a func(str)->int or one of the strings %s' % (Checksum._algos.keys(),))

        def calc_func(x):
            return func(x.bytes) & 0xffffffff

        bit_field = BitField(value=0, length=length, encoder=encoder)
        super(Checksum, self).__init__(depends_on=depends_on, bit_field=bit_field, calc_func=calc_func, fuzzable=fuzzable, name=name)

def render(self, ctx=None):
        '''
        :param ctx: rendering context in which the method was called
        :rtype: `Bits`
        :return: rendered value of the container
        '''
        self._initialize()
        render_count = 1
        if ctx is None:
            ctx = RenderContext()
            if self._need_second_pass:
                render_count = 2
        ctx.push(self)
        if self.is_default():
            self._current_rendered = self._default_rendered
        else:
            if self.offset is None:
                self.offset = 0
            for i in range(render_count):
                offset = self.offset
                rendered = BitArray()
                for field in self._fields:
                    field.set_offset(offset)
                    frendered = field.render(ctx)
                    if not isinstance(frendered, Bits):
                        raise KittyException('the field %s:%s was rendered to type %s, you should probably wrap it with appropriate encoder' % (
                            field.get_name(), type(field), type(frendered)))
                    rendered.append(frendered)
                    offset += len(frendered)
                self.set_current_value(rendered)
        ctx.pop()
        return self._current_rendered