Merge pull request #42 from gmacon/41-write-big-endian

Pass the endianness to all writing locations

Author: rshk

.pre-commit-config.yaml

@@ -12,7 +12,7 @@ repos:
         additional_dependencies:
           - toml
 -   repo: https://github.com/psf/black
-    rev: 20.8b1
+    rev: 22.3.0
     hooks:
     - id: black
 -   repo: https://gitlab.com/pycqa/flake8

CHANGELOG.rst

@@ -1,6 +1,11 @@
 History
 #######
 
+Unreleased
+==========
+
+- Use the explicit endianness everywhere when writing.
+
 v0.1
 ====
 

examples/dump_tcpip_stats.py

@@ -34,7 +34,7 @@ def human_number(num, k=1000):
     assert isinstance(num, int)
     for i, suffix in enumerate(powers):
         if (num < (k ** (i + 1))) or (i == len(powers) - 1):
-            return "{0:d}{1}".format(int(round(num / (k ** i))), suffix)
+            return "{0:d}{1}".format(int(round(num / (k**i))), suffix)
     raise AssertionError("Should never reach this")
 
 

pcapng/blocks.py

@@ -54,7 +54,9 @@ def __init__(self, **kwargs):
             for key, packed_type, default in self.schema:
                 if key == "options":
                     self._decoded[key] = Options(
-                        schema=packed_type.options_schema, data={}, endianness="="
+                        schema=packed_type.options_schema,
+                        data={},
+                        endianness=kwargs["endianness"],
                     )
                     if "options" in kwargs:
                         for oky, ovl in kwargs["options"].items():
@@ -83,10 +85,10 @@ def _write(self, outstream):
         block_length = 12 + subblock_length
         if subblock_length % 4 != 0:
             block_length += 4 - (subblock_length % 4)
-        write_int(self.magic_number, outstream, 32)
-        write_int(block_length, outstream, 32)
+        write_int(self.magic_number, outstream, 32, endianness=self.section.endianness)
+        write_int(block_length, outstream, 32, endianness=self.section.endianness)
         write_bytes_padded(outstream, encoded_block)
-        write_int(block_length, outstream, 32)
+        write_int(block_length, outstream, 32, endianness=self.section.endianness)
 
     def _encode(self, outstream):
         """Encodes the fields of this block into raw data"""

pcapng/flags.py

@@ -84,14 +84,14 @@ def __init__(self, owner, offset, size, extra=None):
                 "{cls} needs an iterable of values".format(cls=self.__class__.__name__)
             )
         extra = list(extra)
-        if len(extra) > 2 ** size:
+        if len(extra) > 2**size:
             raise TypeError(
                 "{cls} iterable has too many values (got {got}, "
                 "{size} bits only address {max})".format(
                     cls=self.__class__.__name__,
                     got=len(extra),
                     size=size,
-                    max=2 ** size,
+                    max=2**size,
                 )
             )
 

tests/test_parse_enhanced_packet.py

@@ -93,7 +93,7 @@ def test_read_block_enhanced_packet_bigendian():
 def _generate_file_with_tsresol(base, exponent):
     tsresol = pack_timestamp_resolution(base, exponent)
     base_timestamp = 1420070400.0  # 2015-01-01 00:00 UTC
-    timestamp = base_timestamp / (base ** exponent)
+    timestamp = base_timestamp / (base**exponent)
 
     stream = io.BytesIO()
 
@@ -167,6 +167,6 @@ def test_read_block_enhanced_packet_tsresol_bigendian(tsr_base, tsr_exp):
     assert blocks[2].interface_id == 0
     assert blocks[2].interface == blocks[1]
 
-    resol = tsr_base ** tsr_exp
+    resol = tsr_base**tsr_exp
     assert blocks[2].timestamp_resolution == resol
     assert blocks[2].timestamp == 1420070400.0

tests/test_parse_wireshark_capture_files.py

@@ -105,7 +105,7 @@ def test_sample_test005_ntar():
             blocks[1].options.get_raw("if_speed") == b"\x00\xe4\x0b\x54\x02\x00\x00\x00"
         )  # noqa
         assert blocks[1].options["if_speed"] == 0x00000002540BE400
-        assert blocks[1].options["if_speed"] == (10 ** 10)  # 10Gbit
+        assert blocks[1].options["if_speed"] == (10**10)  # 10Gbit
 
         assert blocks[1].options["if_description"] == "Stupid ethernet interface\x00"
 
@@ -143,7 +143,7 @@ def test_sample_test006_ntar(filename):
         assert blocks[1].snaplen == 96
         assert len(blocks[1].options) == 2
 
-        assert blocks[1].options["if_speed"] == (10 ** 8)  # 100Mbit
+        assert blocks[1].options["if_speed"] == (10**8)  # 100Mbit
 
         assert blocks[1].options["if_description"] == "Stupid ethernet interface\x00"
 

tests/test_utils.py

@@ -38,9 +38,9 @@ def test_unpack_tsresol():
     assert unpack_timestamp_resolution(bytes((100,))) == 1e-100
 
     assert unpack_timestamp_resolution(bytes((0 | 0b10000000,))) == 1
-    assert unpack_timestamp_resolution(bytes((1 | 0b10000000,))) == 2 ** -1
-    assert unpack_timestamp_resolution(bytes((6 | 0b10000000,))) == 2 ** -6
-    assert unpack_timestamp_resolution(bytes((100 | 0b10000000,))) == 2 ** -100
+    assert unpack_timestamp_resolution(bytes((1 | 0b10000000,))) == 2**-1
+    assert unpack_timestamp_resolution(bytes((6 | 0b10000000,))) == 2**-6
+    assert unpack_timestamp_resolution(bytes((100 | 0b10000000,))) == 2**-100
 
 
 def test_pack_tsresol():

tests/test_write_support.py

@@ -17,17 +17,19 @@ def compare_blocklists(list1, list2):
         assert list1[i] == list2[i], "block #{} mismatch".format(i)
 
 
-def test_write_read_all_blocks():
+@pytest.mark.parametrize("endianness", ["<", ">"])
+def test_write_read_all_blocks(endianness):
     # Track the blocks we're writing
     out_blocks = []
 
     # Build our original/output session
     o_shb = blocks.SectionHeader(
+        endianness=endianness,
         options={
             "shb_hardware": "pytest",
             "shb_os": "python",
             "shb_userappl": "python-pcapng",
-        }
+        },
     )
     out_blocks.append(o_shb)
 
@@ -129,7 +131,8 @@ def test_write_read_all_blocks():
     compare_blocklists(in_blocks, out_blocks)
 
 
-def test_spb_snap_lengths():
+@pytest.mark.parametrize("endianness", ["<", ">"])
+def test_spb_snap_lengths(endianness):
     """
     Simple Packet Blocks present a unique challenge in parsing. The packet does not
     contain an explicit "captured length" indicator, only the original observed
@@ -147,7 +150,7 @@ def test_spb_snap_lengths():
     data = bytes(range(0, 256))
 
     # First session: no snap length
-    o_shb = blocks.SectionHeader()
+    o_shb = blocks.SectionHeader(endianness=endianness)
     o_idb = o_shb.new_member(blocks.InterfaceDescription)  # noqa: F841
     o_blk1 = o_shb.new_member(blocks.SimplePacket, packet_data=data)
 
@@ -162,7 +165,7 @@ def test_spb_snap_lengths():
     assert i_blk1.packet_data == data
 
     # Second session: with snap length
-    o_shb = blocks.SectionHeader()
+    o_shb = blocks.SectionHeader(endianness=endianness)
     o_idb = o_shb.new_member(blocks.InterfaceDescription, snaplen=32)  # noqa: F841
     o_blk1 = o_shb.new_member(blocks.SimplePacket, packet_data=data[:16])
     o_blk2 = o_shb.new_member(blocks.SimplePacket, packet_data=data[:32])