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.
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=b'', context_data=b'')¶ 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
toNone
even if it works.- Parameters
- Returns
the number of bytes that were decoded.
- Return type
-
encode
(full_data=b'', context_data=b'')¶ Return the encoded binary string.
-
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 withpadding
. If set to acallable
(lambda or function) the result is a fixed or variable length string padded withpadding
.
-
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=b'', context_data=b'')¶ Decode a binary string and return the number of bytes that were decoded.
- Parameters
- Returns
the number of bytes that were decoded.
- Return type
-
encode
(full_data=b'', context_data=b'')¶ Return the encoded binary string.
-
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 withpadding
. If set to acallable
(lambda or function) the result is a fixed or variable length string padded withpadding
.
-
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=b'', padding=b'\x00')¶ RegularExpressionMatch
creates aoser.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) andcontext_data
the data within the current hierarchy. Ifstop_condition
returnsTrue
decoding stops.prototype (an OSER class) – a class that is the prototype for values
stop_on_exception (bool) – if set to
True
aDecodeException
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 ofprototype(*args, **kwargs)
.
-
byte_size
()¶ Return the current size in bytes.
-
decode
(data, full_data=b'', context_data=b'')¶ Decode a binary string and return the number of bytes that were decoded.
- Parameters
- Returns
the number of bytes that were decoded.
- Return type
-
encode
(full_data=b'', context_data=b'')¶ Return the encoded binary string.
-
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 toFalse
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
-
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)