Merge pull request #13 from ronfrenk/TLV_length

Implemented reading of the length of a TLV structure

Author: russss

emv/protocol/data.py

@@ -54,6 +54,24 @@ def read_tag(data):
     return tag, i
 
 
+def read_length(data):
+    """ Read length from a list of bytes, starting at the first byte.
+        Returns the length, plus the number of bytes read from the list.
+
+        EMV 4.3 Book 3 Annex B2
+    """
+    i = 0
+    length = data[i]
+    i += 1
+    if length & 0x80:
+        length_bytes_count = length & 0x7F
+        length = 0
+        for j in range(length_bytes_count):
+            length = (length << 8) + data[i + j]
+        i += length_bytes_count
+    return length, i
+
+
 @total_ordering
 class Tag(object):
     """ Represents a data tag. Provides ordering and pretty rendering."""

emv/protocol/structures.py

@@ -1,5 +1,12 @@
 from collections import OrderedDict
-from .data import ELEMENT_FORMAT, render_element, read_tag, is_constructed, Tag
+from .data import (
+    ELEMENT_FORMAT,
+    render_element,
+    read_tag,
+    read_length,
+    is_constructed,
+    Tag,
+)
 from .data_elements import Parse, EPC_PRODUCT_ID
 from ..util import decode_int, bit_set
 import logging
@@ -45,8 +52,10 @@ def unmarshal(cls, data):
             if len(data) <= i:
                 log.info("Invalid TLV - read beyond end of buffer at %s: %s", tag, data)
                 return data
-            length = data[i]
-            i += 1
+
+            length, length_len = read_length(data[i:])
+            i += length_len
+
             value = data[i : i + length]
 
             if is_constructed(tag[0]):

emv/protocol/test/test_structures.py

@@ -31,6 +31,23 @@ def test_tlv(self):
         tlv = TLV.unmarshal(data)
         self.assertEqual(tlv[(0x9F, 0x17)][0], 3)
 
+    def test_length_parsing(self):
+        data = unformat_bytes("42 01 03")
+        tlv = TLV.unmarshal(data)
+        self.assertEqual(tlv[0x42][0], 3)
+
+        data = unformat_bytes("42 81 01 07")
+        tlv = TLV.unmarshal(data)
+        self.assertEqual(tlv[0x42][0], 7)
+
+        data = unformat_bytes("42 82 00 01 07")
+        tlv = TLV.unmarshal(data)
+        self.assertEqual(tlv[0x42][0], 7)
+
+        data = unformat_bytes("42 83 00 00 01 07")
+        tlv = TLV.unmarshal(data)
+        self.assertEqual(tlv[0x42][0], 7)
+
     def test_duplicate_tags(self):
         # An ADF entry with a number of records:
         data = unformat_bytes(