Special types

Constant

You can make any value a constant using oser.Constant().

oser.Constant() raises a ValueError if the current value does not match the desired value.

oser.Constant(*args, **kwargs)

Wrapper to make any OSER type or class a constant.

Parameters:
  • *args (tuple) – the first positional argument is the class that will be made constant. All other parameters are passed to the __init__ of the first positional argument.
  • **kwargs (dict) – keyword parameters are passed to the __init__ of the first positional argument.

Usage:

>>> from oser import Constant
>>> from oser import ULInt8
>>> from oser import ByteStruct
>>> from oser import String
>>> from oser import to_hex
>>>
>>> c = Constant(ULInt8, 23) # constant scalar
>>> print(c.introspect())
   0 \x17  23 (ULInt8)

>>> binary = c.encode()
>>> print(to_hex(binary))
   0
\x17
>>> c.decode("\x00")
ValueError: Constant value not matched: expected value: 23, decoded value: 0
>>>
>>> class Foo(ByteStruct): # constant scalar nested in a ByteStruct
...     def __init__(self):
...         super(Foo, self).__init__()
...         self.c = Constant(ULInt8, 23)
...
>>> f = Foo()
>>> print(f.introspect())
   -    -  Foo():
   0 \x17      c: 23 (ULInt8)

>>> binary = f.encode()
>>> print(to_hex(binary))
   0
\x17
>>> f.decode("\x00")
ValueError: Constant value not matched: expected value: 23, decoded value: 0
>>>
>>> c = Constant(String, length=10, value="abcdefghij") # constant String
>>> print(c.introspect())
   -    -  String():
   0 \x61      'a'
   1 \x62      'b'
   2 \x63      'c'
   3 \x64      'd'
   4 \x65      'e'
   5 \x66      'f'
   6 \x67      'g'
   7 \x68      'h'
   8 \x69      'i'
   9 \x6a      'j'

>>> binary = c.encode()
>>> print(to_hex(binary))
   0|  1|  2|  3|  4|  5|  6|  7|  8|  9
\x61\x62\x63\x64\x65\x66\x67\x68\x69\x6A
>>> c.decode("\x61\x62\x63\x64\x65\x66\x67\x68\x69B")
ValueError: Constant value not matched: expected value: abcdefghij, decoded value: abcdefghiB

Google Protocol Buffers Adapter

Google Protocol Buffers can be used with an adapter. This way it is possible to create protocols that include messages using Google Protocol Buffers but can also be extended with a CRC for example.

class oser.ProtocolBuffersAdapter(instance=None, length=None)

ProtocolBuffersAdapter builds a Google Protocol Buffers adapter.

Parameters:
  • instance=None – an instance of a Google Protocol Buffers class.
  • length=None – the length of the serialized data. If None the rest of the data is used so in most cases it should be a lambda expression. Look at the examples for more details.
byte_size()

Return the size in bytes.

decode(data, full_data='', context_data='')

Decode a binary string and return the number of bytes that were decoded.

If length is None, decoding is done iterative starting with the full buffer and removing one last byte after another until the buffer can be decoded. The buffer must fit the decoded message in protobuf and must not be longer or shorter.

If decoding is not possible at all a DecodeException is raised.

It is recommended not to set length to None even if it works.

Parameters:
  • data (bytes) – the data buffer that is decoded.
  • full_data (bytes) – the binary data string until the part to be decoded. The user normally does not need to supply this.
  • context_data (bytes) – the binary data of the current context. The user normally does not need to supply this.
Returns:

the number of bytes that were decoded.

Return type:

int

encode(full_data='', context_data='')

Return the encoded binary string.

Parameters:
  • full_data (bytes) – the binary data string until the part to be encoded. The user normally does not need to supply this.
  • context_data (bytes) – the binary data of the current context. The user normally does not need to supply this.
Returns:

the encoded binary string.

Return type:

bytes

get()

Return the protocol buffers instance.

introspect(stop_at=None)

Return the introspection representation of the object as a string.

root()

return root element

set(instance)

Set the protocol buffers instance.

Parameters:instance (object) – the protocol buffers instance
set_fuzzing_values(values)

Set fuzzing values.

Parameters:values (iterable) – the values used for fuzzing.
set_length(length)

Set the length.

Parameters:length=None – states the string length. Can be a callable (e.g. lambda) or a scalar. If set to None the result is a null-terminated string. If set to an Integer the result is a fixed length string padded with padding. If set to a callable (lambda or function) the result is a fixed or variable length string padded with padding.
size()

Return the size in bytes.

up()

Return the parent element.

The example uses the following proto file:

package tutorial;

message Person {
  required string name = 1;
  required int32 id = 2;
  optional string email = 3;

  enum PhoneType {
    MOBILE = 0;
    HOME = 1;
    WORK = 2;
  }

  message PhoneNumber {
    required string number = 1;
    optional PhoneType type = 2 [default = HOME];
  }

  repeated PhoneNumber phone = 4;
}

message AddressBook {
  repeated Person person = 1;
}

Usage:

>>> from oser import ByteStruct
>>> from oser import UBInt16
>>> from oser import CRCB32
>>> from oser import to_hex
>>> from oser import ProtocolBuffersAdapter
>>> from tutorial_pb2 import Person
>>>
>>> class Data(ByteStruct):
...         def __init__(self):
...             super(Data, self).__init__()
...
...             self.payloadLength = UBInt16(0)
...             person = Person()
...             person.id = 1
...             person.name = "name"
...             person.PhoneNumber.number = 2
...             self.payload = ProtocolBuffersAdapter(object=person, length=lambda self: self.payloadLength.get())
...             self.crc = CRCB32(strict=True)
...
...         def encode(self, full_data="", context_data=""):#
...             self.payloadLength.set(len(self.payload.encode()))
...             #print("set length to", self.length.get())
...             return super(Data, self).encode()
...
>>> instance = Data()
>>>
>>> binary = instance.encode()
>>> print(instance)
Data():
    payloadLength: 8 (UBInt16)
    payload:
name
    crc: 3809365856 (CRCB32)

>>> print(instance.introspect())
   -    -  Data():
   0 \x00      payloadLength: 8 (UBInt16)
   1 \x08
   -    -      payload: ProtocolBuffersAdapter(Person):
   2 \x0a          '\n'
   3 \x04          '\x04'
   4 \x6e          'n'
   5 \x61          'a'
   6 \x6d          'm'
   7 \x65          'e'
   8 \x10          '\x10'
   9 \x01          '\x01'
  10 \xe3      crc: 3809365856 (CRCB32)
  11 \x0e
  12 \x4f
  13 \x60

>>> print(to_hex(binary))
   0|  1|  2|  3|  4|  5|  6|  7|  8|  9| 10| 11| 12| 13
\x00\x08\x0A\x04\x6E\x61\x6D\x65\x10\x01\xE3\x0E\x4F\x60
>>>
>>> # alter data
... person = instance.payload.get()
>>> person.id = 123456789
>>> person.name = "another name"
>>> person.PhoneNumber.number = 1337
>>>
>>> binary = instance.encode()
>>> print(instance)
Data():
    payloadLength: 19 (UBInt16)
    payload:

another name���:
    crc: 3147422169 (CRCB32)

>>> print(instance.introspect())
   -    -  Data():
   0 \x00      payloadLength: 19 (UBInt16)
   1 \x13
   -    -      payload: ProtocolBuffersAdapter(Person):
   2 \x0a          '\n'
   3 \x0c          '\x0c'
   4 \x61          'a'
   5 \x6e          'n'
   6 \x6f          'o'
   7 \x74          't'
   8 \x68          'h'
   9 \x65          'e'
  10 \x72          'r'
  11 \x20          ' '
  12 \x6e          'n'
  13 \x61          'a'
  14 \x6d          'm'
  15 \x65          'e'
  16 \x10          '\x10'
  17 \x95          '\x95'
  18 \x9a          '\x9a'
  19 \xef          '\xef'
  20 \x3a          ':'
  21 \xbb      crc: 3147422169 (CRCB32)
  22 \x99
  23 \xd9
  24 \xd9

>>> print(to_hex(binary))
   0|  1|  2|  3|  4|  5|  6|  7|  8|  9| 10| 11| 12| 13| 14| 15| 16| 17| 18| 19| 20| 21| 22| 23| 24
\x00\x13\x0A\x0C\x61\x6E\x6F\x74\x68\x65\x72\x20\x6E\x61\x6D\x65\x10\x95\x9A\xEF\x3A\xBB\x99\xD9\xD9
>>>
>>> bytesDecoded = instance.decode("\x00\x10\x0A\x0C\x48\x49\x4C\x53\x54\x45\x52\x20\x47\x6D\x62\x48\x10\x01\x0F\x5B\x58\x85")
>>> print(bytesDecoded)
22
>>> print(instance)
Data():
    payloadLength: 16 (UBInt16)
    payload:

HILSTER GmbH
    crc: 257644677 (CRCB32)

>>> print(instance.introspect())
   -    -  Data():
   0 \x00      payloadLength: 16 (UBInt16)
   1 \x10
   -    -      payload: ProtocolBuffersAdapter(Person):
   2 \x0a          '\n'
   3 \x0c          '\x0c'
   4 \x48          'H'
   5 \x49          'I'
   6 \x4c          'L'
   7 \x53          'S'
   8 \x54          'T'
   9 \x45          'E'
  10 \x52          'R'
  11 \x20          ' '
  12 \x47          'G'
  13 \x6d          'm'
  14 \x62          'b'
  15 \x48          'H'
  16 \x10          '\x10'
  17 \x01          '\x01'
  18 \x0f      crc: 257644677 (CRCB32)
  19 \x5b
  20 \x58
  21 \x85

>>>
>>> print("name:", instance.payload.get().name)
name: HILSTER GmbH

JSON Adapter

It is possible to serialize and deserialize JSON with OSER.

class oser.JSONAdapter(data=None)

JSONAdapter builds a JSON serializer.

Parameters:data=None – the data to be json-ized.
byte_size()

Return the size in bytes.

decode(data, full_data='', context_data='')

Decode a binary string and return the number of bytes that were decoded.

Parameters:
  • data (bytes) – the data buffer that is decoded.
  • full_data (bytes) – the binary data string until the part to be decoded. The user normally does not need to supply this.
  • context_data (bytes) – the binary data of the current context. The user normally does not need to supply this.
Returns:

the number of bytes that were decoded.

Return type:

int

encode(full_data='', context_data='')

Return the encoded binary string.

Parameters:
  • full_data (bytes) – the binary data string until the part to be encoded. The user normally does not need to supply this.
  • context_data (bytes) – the binary data of the current context. The user normally does not need to supply this.
Returns:

the encoded binary string.

Return type:

bytes

get()

Return the value.

Returns:the value
introspect(stop_at=None)

Return the introspection representation of the object as a string.

root()

return root element

set(value)

Set the value.

Parameters:value – the new value
set_fuzzing_values(values)

Set fuzzing values.

Parameters:values (iterable) – the values used for fuzzing.
set_length(length)

Set the length.

Parameters:length=None – states the string length. Can be a callable (e.g. lambda) or a scalar. If set to None the result is a null-terminated string. If set to an Integer the result is a fixed length string padded with padding. If set to a callable (lambda or function) the result is a fixed or variable length string padded with padding.
size()

Return the size in bytes.

up()

Return the parent element.

Usage:

>>> from oser import JSONAdapter
>>> data = {
...     "list" : [ii for ii in range(3)],
...     "scalar" : 1337,
...     "string" : "test"
...     }

>>> j = JSONAdapter(data=data)
>>> print(j)
{'scalar': 1337, 'list': [0, 1, 2], 'string': 'test'}

>>> print(j.encode())
{"list":[0,1,2],"scalar":1337,"string":"test"}
>>> print(j.introspect())
   -    -  JSONAdapter():
   0 \x7b      '{'
   1 \x22      '"'
   2 \x6c      'l'
   3 \x69      'i'
   4 \x73      's'
   5 \x74      't'
   6 \x22      '"'
   7 \x3a      ':'
   8 \x5b      '['
   9 \x30      '0'
  10 \x2c      ','
  11 \x31      '1'
  12 \x2c      ','
  13 \x32      '2'
  14 \x5d      ']'
  15 \x2c      ','
  16 \x22      '"'
  17 \x73      's'
  18 \x63      'c'
  19 \x61      'a'
  20 \x6c      'l'
  21 \x61      'a'
  22 \x72      'r'
  23 \x22      '"'
  24 \x3a      ':'
  25 \x31      '1'
  26 \x33      '3'
  27 \x33      '3'
  28 \x37      '7'
  29 \x2c      ','
  30 \x22      '"'
  31 \x73      's'
  32 \x74      't'
  33 \x72      'r'
  34 \x69      'i'
  35 \x6e      'n'
  36 \x67      'g'
  37 \x22      '"'
  38 \x3a      ':'
  39 \x22      '"'
  40 \x74      't'
  41 \x65      'e'
  42 \x73      's'
  43 \x74      't'
  44 \x22      '"'
  45 \x7d      '}'
  46 \x00      '\x00'
>>>
>>> j2 = JSONAdapter()
>>> j2.decode("[1,2,3]\x00")
8
>>> print(j2)
[1, 2, 3]
>>>
>>> print(j2.get())
[1, 2, 3]
>>> print(type(j2.get()))
<type 'list'>
>>> j3 = JSONAdapter()
>>> j3.set([4,5,6])
>>> print(j3)

Regular Expression Match

If you need a oser.String that fulfills some constraints you can use oser.RegularExpressionMatch() to do this. oser.RegularExpressionMatch() expects a regular expression that is applied to the String`s value. A `RegularExpressionMismatchException is raised if the current value does not match the regular expression.

oser.RegularExpressionMatch(pattern, length, value='', padding='\x00')

RegularExpressionMatch creates a oser.String matches a regular expression.

Parameters:
  • pattern – a regular expression pattern.
  • length – the length of the underlying string.
  • value="" – the initial value.
  • padding=b"" – padding character for the string when it is encoded.

Usage (only match numeric strings):

>>> from oser import ByteStruct
>>> from oser import ULInt8
>>> from oser import RegularExpressionMatch
>>> from oser import to_hex
>>>
>>> class Foo(ByteStruct):
...     def __init__(self):
...         super(Foo, self).__init__()
...
...         self.length = ULInt8(10)
...         self.numbers = RegularExpressionMatch(pattern="^[0-9]*$", length=lambda self: self.length.get(), value="0123456789")
...
>>> f = Foo()
>>> print(f.introspect())
   -    -  Foo():
   0 \x0a      length: 10 (ULInt8)
   -    -      numbers: RegularExpressionMatch():
   1 \x30          '0'
   2 \x31          '1'
   3 \x32          '2'
   4 \x33          '3'
   5 \x34          '4'
   6 \x35          '5'
   7 \x36          '6'
   8 \x37          '7'
   9 \x38          '8'
  10 \x39          '9'

>>> binary = f.encode()
>>> print(to_hex(binary))
   0|  1|  2|  3|  4|  5|  6|  7|  8|  9| 10
\x0A\x30\x31\x32\x33\x34\x35\x36\x37\x38\x39
>>> f.decode("\x0A\x30\x31\x32\x33\x34\x35\x36\x37\x38A")
oser.RegularExpressionMismatchException: Current value (012345678A) does not match regular expression (^[0-9]*$)

Repeat Until

Sometimes you have optional data in your data stream or you need to decode items until the end of a stream or until a stop condition is met. Then oser.Array might not be an option if you don’t know the number of items.

In this cases oser.RepeatUntil can help. In general it works like oser.Array but length is not a callable. For encoding you can set the items and the length and for decoding the length is determined dynamically while oser.RepeatUntil consumes items until stop_condition returns True, an exception is raised or the data stream ends.

class oser.RepeatUntil(length, stop_condition, prototype, stop_on_exception=False, args=None, kwargs=None, values=None)

RepeatUntil builds an array serializer with a variable length that is determined dynamically on decoding.

Parameters:
  • length (integer) – states the initial length
  • stop_condition (callable) – the stop condition called with context, data (the next bytes to be parsed), full_data (all data) and context_data the data within the current hierarchy. If stop_condition returns True decoding stops.
  • prototype (an OSER class) – a class that is the prototype for values
  • stop_on_exception (bool) – if set to True a DecodeException is handled as a stop condition.
  • args=None (tuple) – a tuple of positional arguments that are passed to prototype when it is instantiated when array size is extended. Note: don’t forget a leading comma when passing one value (1,) ((1) is an integer!).
  • kwargs=None (dict) – a dictionary of named arguments that are passed to prototype when it is instantiated when array size is extended.
  • values=None (list containing instances of prototype) – the initial array values. Can be left empty. If empty the array is automatically resized with instances of prototype(*args, **kwargs).
byte_size()

Return the current size in bytes.

decode(data, full_data='', context_data='')

Decode a binary string and return the number of bytes that were decoded.

Parameters:
  • data (bytes) – the data buffer that is decoded.
  • full_data (bytes) – the binary data string until the part to be decoded. The user normally does not need to supply this.
  • context_data (bytes) – the binary data of the current context. The user normally does not need to supply this.
Returns:

the number of bytes that were decoded.

Return type:

int

encode(full_data='', context_data='')

Return the encoded binary string.

Parameters:
  • full_data (bytes) – the binary data string until the part to be encoded. The user normally does not need to supply this.
  • context_data (bytes) – the binary data of the current context. The user normally does not need to supply this.
Returns:

the encoded binary string.

Return type:

bytes

fuzzing_iterator(copy=False)

The fuzzing iterator iterates over all combinations of set fuzzing values (oser.ByteType.set_fuzzing_values()). If no fuzzing values are set the current struct is yielded.

Parameters:copy=False (bool) – if set to True the generated fuzzing values are deep copies of the original. Creating these deep copies is slow. If set to False the original struct is retruned and the generated value must be used immediately since the next generated overwrites the values on the same value.
Yields:the fuzzing combinations.
get()

Return the value.

introspect(stop_at=None)

Return the introspection representation of the object as a string.

Parameters:stop_at=None (object) – stop introspection at stop_at.
root()

return root element

set_length(length)

Set the length.

Parameters:length (int) – the new length
size()

Return the current size in bytes.

up()

return parent element

Usage:

>>> from oser import ByteStruct, RepeatUntil, CRCB32, UBInt8, to_hex
>>> length = 3
>>> class Data(ByteStruct):
...     def __init__(self):
...         super(Data, self).__init__()
...         self.items = RepeatUntil(length,
...                         stop_condition=lambda ctx,
...                             data, full_data,
...                             context_data: len(data) <= 4,
...                         prototype=UBInt8,
...                         values=[UBInt8(ii) for ii in range(length)])
...         self.crc = CRCB32(strict=True)
...
>>> data = Data()
>>> data.items.set_length(20) # resize
>>> for ii in range(20):
...     data.items[ii].set(ii)
...
>>> encoded = data.encode()
>>> print(to_hex(encoded))
   0|  1|  2|  3|  4|  5|  6|  7|  8|  9| 10| 11| 12| 13| 14| 15| 16| 17| 18| 19| 20| 21| 22| 23
\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0A\x0B\x0C\x0D\x0E\x0F\x10\x11\x12\x13\x6D\xA9\x2B\x61
>>> print(data.introspect())
   -    -  Data():
   -    -      items: RepeatUntil():
   -    -      [
   0 \x00          @0: 0 (UBInt8)
   1 \x01          @1: 1 (UBInt8)
   2 \x02          @2: 2 (UBInt8)
   3 \x03          @3: 3 (UBInt8)
   4 \x04          @4: 4 (UBInt8)
   5 \x05          @5: 5 (UBInt8)
   6 \x06          @6: 6 (UBInt8)
   7 \x07          @7: 7 (UBInt8)
   8 \x08          @8: 8 (UBInt8)
   9 \x09          @9: 9 (UBInt8)
  10 \x0a          @10: 10 (UBInt8)
  11 \x0b          @11: 11 (UBInt8)
  12 \x0c          @12: 12 (UBInt8)
  13 \x0d          @13: 13 (UBInt8)
  14 \x0e          @14: 14 (UBInt8)
  15 \x0f          @15: 15 (UBInt8)
  16 \x10          @16: 16 (UBInt8)
  17 \x11          @17: 17 (UBInt8)
  18 \x12          @18: 18 (UBInt8)
  19 \x13          @19: 19 (UBInt8)
   -    -      ]
  20 \x6d      crc: 1839803233 (CRCB32)
  21 \xa9
  22 \x2b
  23 \x61

>>> b = b"\x00\x01\x02\x03\xC1\xE9\xBD\xED"
>>> data2 = Data()
>>> decoded = data2.decode(b)
>>> print(data2.introspect())
   -    -  Data():
   -    -      items: RepeatUntil():
   -    -      [
   0 \x00          @0: 0 (UBInt8)
   1 \x01          @1: 1 (UBInt8)
   2 \x02          @2: 2 (UBInt8)
   3 \x03          @3: 3 (UBInt8)
   -    -      ]
   4 \xc1      crc: 3253321197 (CRCB32)
   5 \xe9
   6 \xbd
   7 \xed

Lazy initialization

Sometimes you need to speed up some things. You can use the oser.Lazy element to initialize an object on first demand. So if an object is never used it is never initialized saving memory and cpu time.

class oser.Lazy(cls, *args, **kwargs)

Lazy can be used to realize lazy-initialization automatically.

cls is instantiated on first access but not in __init__(). So if no access is done it will never be instantiated and will not occupy any memory and cpu time.

Parameters:
  • cls – the class to be instantiated.
  • *args – a tuple for positional arguments for cls.
  • **kwargs – a dict for named arguments for cls.

Usage:

>>> from oser import Lazy
>>> from oser import ULInt8
>>> from oser import ByteStruct
>>>
>>> class Foo(ByteStruct):
...     def __init__(self):
...         super(Foo, self).__init__()
...
...         self.a = Lazy(ULInt8, 2)
...
>>> foo = Foo()
>>> print(foo) # foo.a is now initialized
Foo():
    a: 2 (ULInt8)