dexter has uploaded this change for review.

View Change

construct: add length steps to StripTrailerAdapter

The class StripTrailerAdapter allows to remove trailing bytes that
match a specified value from the encoding result of a sub-construct.
The result is always the shortest possible remainder of bytes that
do not match the secified value.

Unfortunately there are specifications that explicitly require the
length of the result to fit into a limited set of possible length
values. For example: A result may be either one byte or three byte
long. To cover those cases as well, let's add an array parameter
where we can configure the allowed length values of the encoding
result.

Related: OS#6679
Change-Id: I86df064fa41db85923eeb0d83cc399504fdd4488
---
M src/osmocom/construct.py
M tests/test_construct.py
2 files changed, 41 insertions(+), 3 deletions(-)

git pull ssh://gerrit.osmocom.org:29418/python/pyosmocom refs/changes/62/39162/1

diff --git a/src/osmocom/construct.py b/src/osmocom/construct.py
index c69b44f..1f5bd45 100644
--- a/src/osmocom/construct.py
+++ b/src/osmocom/construct.py
@@ -413,16 +413,20 @@
     Encoder removes all trailing bytes matching the default_value
     Decoder pads input data up to total_length with default_value
 
+    In case the encoding restricts the length of the result to specific values, the API user may set those restrictions
+    using the steps parameter. (e.g. encoded result must be either 1 or 3 byte long, steps would be set to [1,3])
+
     This is used in constellations like "FlagsEnum(StripTrailerAdapter(GreedyBytes, 3), ..."
     where you have a bit-mask that may have 1, 2 or 3 bytes, depending on whether or not any
     of the LSBs are actually set.
     """
-    def __init__(self, subcon, total_length:int, default_value=b'\x00', min_len=1):
+    def __init__(self, subcon, total_length:int, default_value=b'\x00', min_len=1, steps:[int]=[]):
         super().__init__(subcon)
         assert len(default_value) == 1
         self.total_length = total_length
         self.default_value = default_value
         self.min_len = min_len
+        self.steps = steps
 
     def _decode(self, obj, context, path):
         assert isinstance(obj, bytes)
@@ -435,9 +439,17 @@
         assert isinstance(obj, int)
         obj = obj.to_bytes(self.total_length, 'big')
         # remove trailing bytes if they are zero
+
+        obj_step_aligned = obj
         while len(obj) > self.min_len and obj[-1] == self.default_value[0]:
             obj = obj[:-1]
-        return obj
+            if len(obj) in self.steps:
+                obj_step_aligned = obj
+
+        if self.steps == []:
+            return obj
+        else:
+            return obj_step_aligned;
 
 
 def filter_dict(d, exclude_prefix='_'):
diff --git a/tests/test_construct.py b/tests/test_construct.py
index bcb2cf8..e4234db 100755
--- a/tests/test_construct.py
+++ b/tests/test_construct.py
@@ -90,13 +90,39 @@
                                           final_application=0x0200, global_service=0x0100,
                                           receipt_generation=0x80, ciphered_load_file_data_block=0x40,
                                           contactless_activation=0x20, contactless_self_activation=0x10)
+    PrivilegesSteps = FlagsEnum(StripTrailerAdapter(GreedyBytes, 3, steps = [1,3]), security_domain=0x800000,
+                                          dap_verification=0x400000,
+                                          delegated_management=0x200000, card_lock=0x100000,
+                                          card_terminate=0x080000, card_reset=0x040000,
+                                          cvm_management=0x020000, mandated_dap_verification=0x010000,
+                                          trusted_path=0x8000, authorized_management=0x4000,
+                                          token_management=0x2000, global_delete=0x1000,
+                                          global_lock=0x0800, global_registry=0x0400,
+                                          final_application=0x0200, global_service=0x0100,
+                                          receipt_generation=0x80, ciphered_load_file_data_block=0x40,
+                                          contactless_activation=0x20, contactless_self_activation=0x10)
     examples = ['00', '80', '8040', '400010']
-    def test_examples(self):
+    def test_encdec(self):
         for e in self.examples:
             dec = self.Privileges.parse(h2b(e))
             reenc = self.Privileges.build(dec)
             self.assertEqual(e, b2h(reenc))
 
+    def test_enc(self):
+        enc = self.Privileges.build({'dap_verification' : True})
+        self.assertEqual(b2h(enc), '40')
+        enc = self.Privileges.build({'dap_verification' : True, 'global_service' : True})
+        self.assertEqual(b2h(enc), '4001')
+        enc = self.Privileges.build({'dap_verification' : True, 'global_service' : True, 'contactless_self_activation' : True})
+        self.assertEqual(b2h(enc), '400110')
+
+        enc = self.PrivilegesSteps.build({'dap_verification' : True})
+        self.assertEqual(b2h(enc), '40')
+        enc = self.PrivilegesSteps.build({'dap_verification' : True, 'global_service' : True})
+        self.assertEqual(b2h(enc), '400100')
+        enc = self.PrivilegesSteps.build({'dap_verification' : True, 'global_service' : True, 'contactless_self_activation' : True})
+        self.assertEqual(b2h(enc), '400110')
+
 
 class TestAdapters(unittest.TestCase):
     def test_dns_adapter(self):

To view, visit change 39162. To unsubscribe, or for help writing mail filters, visit settings.