implement support for negative padding

src/bitstruct/__init__.py

@@ -193,7 +193,7 @@ def _parse_format(fmt, names, text_encoding, text_errors):
 
         if names is None:
             name = i
-        elif type_ not in 'pP':
+        elif type_ not in 'pPq':
             name = names[i]
 
         if type_ == 's':
@@ -218,6 +218,8 @@ def _parse_format(fmt, names, text_encoding, text_errors):
             info = _ZeroPadding(size)
         elif type_ == 'P':
             info = _OnePadding(size)
+        elif type_ == 'q':
+            info = _ZeroPadding(-size)
         else:
             raise Error(f"bad char '{type_}' in format")
 
@@ -251,7 +253,18 @@ def __init__(self,
         infos, byte_order = _parse_format(fmt, names, text_encoding, text_errors)
         self._infos = infos
         self._byte_order = byte_order
-        self._number_of_bits_to_unpack = sum([info.size for info in infos])
+        self._number_of_bits_to_unpack = 0
+
+        offset = 0
+        for info in infos:
+            offset += info.size
+
+            if offset < 0:
+                raise Error(f"negative padding points to before start of data in '{fmt}'")
+
+            self._number_of_bits_to_unpack = max(
+                self._number_of_bits_to_unpack,
+                offset)
 
     def pack_value(self, info, value, bits):
         value_bits = info.pack(value)
@@ -277,12 +290,19 @@ def pack_value(self, info, value, bits):
 
     def pack_any(self, values):
         bits = ''
+        offset = 0
 
         for info in self._infos:
             if isinstance(info, _Padding):
-                bits += info.pack()
+                value_bits = '' if info.size < 0 else info.pack()
             else:
-                bits = self.pack_value(info, values[info.name], bits)
+                value_bits = self.pack_value(
+                    info,
+                    values[info.name],
+                    bits[:offset])[offset:]
+
+            bits = bits[:offset] + value_bits + bits[offset + len(value_bits):]
+            offset += info.size
 
         # Padding of last byte.
         tail = len(bits) % 8
@@ -343,15 +363,24 @@ def pack_into_any(self, buf, offset, data, **kwargs):
         fill_padding = kwargs.get('fill_padding', True)
         buf_bits = _pack_bytearray(8 * len(buf), buf)
         bits = buf_bits[0:offset]
+        pos = offset
 
         for info in self._infos:
             if isinstance(info, _Padding):
-                if fill_padding:
-                    bits += info.pack()
+                if info.size < 0:
+                    value_bits = ''
+                elif fill_padding:
+                    value_bits = info.pack()
                 else:
-                    bits += buf_bits[len(bits):len(bits) + info.size]
+                    value_bits = buf_bits[pos:pos + info.size]
             else:
-                bits = self.pack_value(info, data[info.name], bits)
+                value_bits = self.pack_value(
+                    info,
+                    data[info.name],
+                    bits[:pos])[pos:]
+
+            bits = bits[:pos] + value_bits + bits[pos + len(value_bits):]
+            pos += info.size
 
         bits += buf_bits[len(bits):]
 
@@ -503,9 +532,15 @@ def pack(fmt, *args):
     - ``r`` -- raw, bytes
     - ``p`` -- padding with zeros, ignore
     - ``P`` -- padding with ones, ignore
+    - ``q`` -- move back, ignore
 
     Length is the number of bits to pack the value into.
 
+    ``q`` moves back that many bits (``q8``). Following fields then
+    overlap the previous bits; when packing they overwrite them. The
+    size of the format is the furthest bit position reached. Moving
+    before the start of the format is an error.
+
     Example format string with default bit and byte ordering:
     ``'u1u3p7s16'``
 

src/bitstruct/c.c

@@ -21,6 +21,13 @@ struct field_info_t {
     pack_field_t pack;
     unpack_field_t unpack;
     int number_of_bits;
+    int start_bit;
+
+    /* The maximum start bit of any field before and including the current
+       one. If there are no overlapping fields (i.e., no negative
+       padding), this is identical to start_bit */
+    int extent_before;
+
     bool is_padding;
     union {
         struct {
@@ -35,6 +42,7 @@ struct field_info_t {
 
 struct info_t {
     int number_of_bits;
+    bool has_overlap;
     int number_of_fields;
     int number_of_non_padding_fields;
     struct field_info_t fields[1];
@@ -809,6 +817,12 @@ static int field_info_init(struct field_info_t *self_p,
         res = field_info_init_one_padding(self_p);
         break;
 
+    case 'q':
+        /* Move back number_of_bits bits. */
+        is_padding = true;
+        res = field_info_init_zero_padding(self_p);
+        break;
+
     default:
         PyErr_Format(PyExc_ValueError, "Bad format field type '%c'.", kind);
         res = -1;
@@ -833,7 +847,7 @@ static int count_number_of_fields(const char *format_p,
         if ((*format_p >= 'A') && (*format_p <= 'z')) {
             count++;
 
-            if ((*format_p == 'p') || (*format_p == 'P')) {
+            if ((*format_p == 'p') || (*format_p == 'P') || (*format_p == 'q')) {
                 (*number_of_padding_fields_p)++;
             }
         }
@@ -871,6 +885,8 @@ const char *parse_field(const char *format_p,
     if (*number_of_bits_p == 0) {
         PyErr_SetString(PyExc_ValueError, "Field of size 0.");
         format_p = NULL;
+    } else if (*kind_p == 'q') {
+        *number_of_bits_p = -*number_of_bits_p;
     }
 
     return (format_p);
@@ -886,6 +902,8 @@ static struct info_t *parse_format(PyObject *format_obj_p)
     int number_of_bits;
     int number_of_padding_fields;
     int res;
+    int offset;
+    int extent;
 
     format_p = PyUnicode_AsUTF8(format_obj_p);
 
@@ -904,6 +922,9 @@ static struct info_t *parse_format(PyObject *format_obj_p)
     }
 
     info_p->number_of_bits = 0;
+    info_p->has_overlap = false;
+    offset = 0;
+    extent = 0;
     info_p->number_of_fields = number_of_fields;
     info_p->number_of_non_padding_fields = (
         number_of_fields - number_of_padding_fields);
@@ -925,19 +946,105 @@ static struct info_t *parse_format(PyObject *format_obj_p)
             return (NULL);
         }
 
-        if (number_of_bits > (INT_MAX - info_p->number_of_bits)) {
+        if (number_of_bits > (INT_MAX - offset)) {
             PyErr_SetString(PyExc_ValueError, "Format too long.");
             PyMem_RawFree(info_p);
 
             return (NULL);
         }
 
-        info_p->number_of_bits += number_of_bits;
+        info_p->fields[i].start_bit = offset;
+        info_p->fields[i].extent_before = extent;
+
+        if (offset < extent) {
+            info_p->has_overlap = true;
+        }
+
+        offset += number_of_bits;
+
+        if (offset < 0) {
+            PyErr_SetString(PyExc_ValueError,
+                            "Negative padding before start of format.");
+            PyMem_RawFree(info_p);
+
+            return (NULL);
+        }
+
+        if (offset > extent) {
+            extent = offset;
+        }
     }
 
+    info_p->number_of_bits = extent;
+
     return (info_p);
 }
 
+static void pack_field(struct bitstream_writer_t *writer_p,
+                       PyObject *value_p,
+                       struct field_info_t *field_p)
+{
+    int first_idx;
+    int last_idx;
+    int length;
+    int start_bit;
+    int end_bit;
+    uint8_t *buf_p;
+    uint8_t old_first;
+    uint8_t old_last;
+    uint8_t first_mask;
+    uint8_t last_mask;
+
+    if (field_p->is_padding && (field_p->number_of_bits < 0)) {
+        /* deal with negative padding, i.e., go backwards */
+        bitstream_writer_seek(writer_p, field_p->number_of_bits);
+
+        return;
+    }
+
+    if (field_p->start_bit >= field_p->extent_before) {
+        /* append new data to the write buffer */
+        field_p->pack(writer_p, value_p, field_p);
+
+        return;
+    }
+
+    /* The current field starts before the end of the output stream. The
+     * buffer is already allocated (and zeroed) for the whole format.
+     * Overwrite already written bits for the current field while making
+     * sure to not overwrite data belonging to others.
+     */
+    start_bit = bitstream_writer_size_in_bits(writer_p);
+    end_bit = (start_bit + field_p->number_of_bits);
+    first_idx = (start_bit / 8);
+    last_idx = ((end_bit - 1) / 8);
+    length = (last_idx - first_idx + 1);
+
+    first_mask = (0xff >> (start_bit % 8));
+    last_mask = ((0xff00 >> (((end_bit - 1) % 8) + 1)) & 0xff);
+
+    buf_p = writer_p->buf_p;
+    old_first = buf_p[first_idx];
+    old_last = buf_p[last_idx];
+
+    memset(buf_p + first_idx, 0, length);
+    field_p->pack(writer_p, value_p, field_p);
+
+    /* restore the non-field bits of the first and last bytes affected
+     * by the current field */
+    if (first_idx == last_idx) {
+      first_mask |= last_mask;
+      buf_p[first_idx] = ((old_first & ~first_mask)
+			  | (buf_p[first_idx] & first_mask));
+    }
+    else {
+      buf_p[first_idx] = ((old_first & ~first_mask)
+			  | (buf_p[first_idx] & first_mask));
+      buf_p[last_idx] = ((old_last & ~last_mask)
+			 | (buf_p[last_idx] & last_mask));
+    }
+}
+
 static void pack_pack(struct info_t *info_p,
                       PyObject *args_p,
                       int consumed_args,
@@ -957,7 +1064,7 @@ static void pack_pack(struct info_t *info_p,
             consumed_args++;
         }
 
-        info_p->fields[i].pack(writer_p, value_p, field_p);
+        pack_field(writer_p, value_p, field_p);
     }
 }
 
@@ -972,6 +1079,10 @@ static PyObject *pack_prepare(struct info_t *info_p,
         return (NULL);
     }
 
+    if (info_p->has_overlap) {
+        memset(PyBytes_AS_STRING(packed_p), 0, PyBytes_GET_SIZE(packed_p));
+    }
+
     bitstream_writer_init(writer_p, (uint8_t *)PyBytes_AS_STRING(packed_p));
 
     return (packed_p);
@@ -1048,7 +1159,6 @@ static PyObject *unpack(struct info_t *info_p,
     PyObject *value_p;
     Py_buffer view = {NULL, NULL};
     int i;
-    int tmp;
     int produced_args;
     int res;
     int allow_truncated;
@@ -1063,14 +1173,16 @@ static PyObject *unpack(struct info_t *info_p,
 
     if (allow_truncated) {
         num_result_fields = 0;
-        tmp = 0;
         for (i = 0; i < info_p->number_of_fields; i++) {
-            if (view.len*8 < tmp + info_p->fields[i].number_of_bits) {
+            if (view.len*8 < (info_p->fields[i].start_bit
+                              + info_p->fields[i].number_of_bits)) {
+                /* input is too short for full unpack. Note that this
+                 * discards all fields after negative padding even if
+                 * they would be inside the limits of the input data.
+                 */
                 break;
             }
 
-            tmp += info_p->fields[i].number_of_bits;
-
             if (!info_p->fields[i].is_padding) {
                 ++num_result_fields;
             }
@@ -1399,7 +1511,7 @@ static void pack_dict_pack(struct info_t *info_p,
             }
         }
 
-        info_p->fields[i].pack(writer_p, value_p, field_p);
+        pack_field(writer_p, value_p, field_p);
     }
 }
 
@@ -1509,7 +1621,12 @@ static PyObject *unpack_dict(struct info_t *info_p,
     produced_args = 0;
 
     for (i = 0; i < info_p->number_of_fields; i++) {
-        if (view.len*8 < reader.bit_offset + info_p->fields[i].number_of_bits)
+        if (view.len*8 < (bitstream_reader_tell(&reader)
+                          + info_p->fields[i].number_of_bits))
+            /* input is too short to unpack the field. all following
+             * fields will be ignored even it they are in the bounds
+             * of the input data due to negative padding.
+             */
             break;
 
         value_p = info_p->fields[i].unpack(&reader, &info_p->fields[i]);

tests/test_bitstruct.py

@@ -497,6 +497,42 @@ def test_byte_order_format(self):
         self.assertEqual(cf.unpack(b''), ())
         self.assertEqual(cf.unpack(b'\x00'), ())
 
+    def test_overlapping_fields(self):
+        self.assertEqual(pack('u8q8u4u4', 0xab, 1, 2), b'\x12')
+        self.assertEqual(pack('u8q4u8', 0xab, 0xcd), b'\xac\xd0')
+        self.assertEqual(pack('u16q12u4q3u3', 0xabcd, 0, 7), b'\xa7\xcd')
+        self.assertEqual(unpack('u8q8u4u4', b'\xab'), (0xab, 0xa, 0xb))
+        self.assertEqual(unpack('u8q4u8', b'\xab\xcd'), (0xab, 0xbc))
+        self.assertEqual(calcsize('u8q8u4'), 8)
+        self.assertEqual(pack_dict('u8q8u4', ['a', 'b'], {'a': 0x12, 'b': 0xf}),
+                         b'\xf2')
+        self.assertEqual(unpack_dict('u8q8u4', ['a', 'b'], b'\xab'),
+                         {'a': 0xab, 'b': 0xa})
+
+        packed = bytearray(b'\xff\xff\xff')
+        pack_into('u8q8u4', packed, 4, 0x12, 0xf)
+        self.assertEqual(packed, b'\xff\x2f\xff')
+
+        packed = bytearray(b'\xff\xff\xff')
+        pack_into('u8q4u8u4', packed, 0, 0, 0, 0)
+        self.assertEqual(packed, b'\x00\x00\xff')
+
+        # Skipped bits are not zeroed.
+        self.assertEqual(pack('u8q4', 0xab), b'\xab')
+        self.assertEqual(pack('u8q4u2', 0xab, 1), b'\xa7')
+        packed = bytearray(b'\xff\xff\xff')
+        pack_into('u8q4', packed, 4, 0xab)
+        self.assertEqual(packed, b'\xfa\xbf\xff')
+
+        with self.assertRaises(Exception):
+            pack('u8q9', 1)
+
+        with self.assertRaises(Exception):
+            pack('u8p-8', 1)
+
+        with self.assertRaises(Exception):
+            pack('u8s-8', 1)
+
     def test_pack_into(self):
         """Pack values into a buffer.