add CRC validation for bluefile metadata

Author: Teque5

sigmf/convert/blue.py

@@ -18,6 +18,7 @@
 import struct
 import tempfile
 from datetime import datetime, timezone
+from functools import lru_cache
 from pathlib import Path
 from typing import Optional, Tuple
 
@@ -62,7 +63,6 @@
 ]
 # fmt: on
 
-HEADER_SIZE_BYTES = 512
 BLOCK_SIZE_BYTES = 512
 
 TYPE_MAP = {
@@ -143,10 +143,9 @@ def detect_endian(data: bytes) -> str:
     endianness = data[8:12].decode("ascii")
     if endianness == "EEEI":
         return "<"
-    elif endianness == "IEEE":
+    if endianness == "IEEE":
         return ">"
-    else:
-        raise SigMFConversionError(f"Unsupported endianness: {endianness}")
+    raise SigMFConversionError(f"Unsupported endianness: {endianness}")
 
 
 def read_hcb(file_path: Path) -> (dict, dict):
@@ -183,8 +182,8 @@ def read_hcb(file_path: Path) -> (dict, dict):
             raw = header_bytes[offset : offset + size]
             try:
                 val = struct.unpack(endian + fmt, raw)[0]
-            except struct.error:
-                raise SigMFConversionError(f"Failed to unpack field {key} with endian {endian}")
+            except struct.error as err:
+                raise SigMFConversionError(f"Failed to unpack field {key} with endian {endian}") from err
             if isinstance(val, bytes):
                 val = val.decode("ascii", errors="replace")
             h_fixed[key] = val
@@ -222,6 +221,20 @@ def read_hcb(file_path: Path) -> (dict, dict):
             raw_adjunct = handle.read(256)
             h_adjunct = {"raw_base64": base64.b64encode(raw_adjunct).decode("ascii")}
 
+        if h_fixed["keywords"].get("CRC") is not None:
+            # CRC calculated on first 160 bytes of fixed header + full extended header
+            handle.seek(0)
+            buffer = handle.read(160)
+            handle.seek(int(h_fixed["ext_start"]) * BLOCK_SIZE_BYTES)
+            buffer += handle.read(int(h_fixed["ext_size"]))
+            target_crc = h_fixed["keywords"]["CRC"].lower()
+            if target_crc == _crc32_broken(buffer):
+                log.debug("CRC ok (BLUE implementation)")
+            elif target_crc == _crc32_posix(buffer):
+                log.debug("CRC ok (POSIX implementation)")
+            else:
+                log.warning("CRC mismatch in BLUE metadata!")
+
     validate_fixed(h_fixed)
     validate_adjunct(h_adjunct)
 
@@ -322,11 +335,6 @@ def data_loopback(blue_path: Path, data_path: Path, h_fixed: dict) -> None:
         Destination path for the SigMF dataset (.sigmf-data).
     h_fixed : dict
         Header Control Block dictionary.
-
-    Returns
-    -------
-    numpy.ndarray
-        Parsed samples. Empty array for zero-sample files.
     """
     header_bytes, data_bytes, _ = _get_blue_boundaries(blue_path, h_fixed)
     blue_format = h_fixed.get("format")
@@ -371,6 +379,70 @@ def data_loopback(blue_path: Path, data_path: Path, h_fixed: dict) -> None:
     log.info("wrote SigMF dataset to %s", data_path)
 
 
+@lru_cache
+def _generate_crc_table(poly: int):
+    """generate lookup table for given polynomial"""
+    table = []
+    for idx in range(256):
+        crc = idx << 24
+        for _ in range(8):
+            if crc & 0x80000000:
+                crc = (crc << 1) ^ poly
+            else:
+                crc = crc << 1
+            crc &= 0xFFFFFFFF
+        table.append(crc)
+    return table
+
+
+def _crc32_posix(data: bytes) -> str:
+    """
+    POSIX.2 CRC-32 with buffer length included
+
+    Supposed BLUE standard implementation.
+    """
+    table = _generate_crc_table(0x04C11DB7)
+    crc = 0
+
+    # process data bytes
+    for byte_val in data:
+        crc = (crc << 8) ^ table[(crc >> 24) ^ byte_val]
+        crc &= 0xFFFFFFFF
+
+    # include buffer size in calculation
+    size_val = len(data)
+    while size_val > 0:
+        crc = (crc << 8) ^ table[(crc >> 24) ^ (size_val & 0xFF)]
+        crc &= 0xFFFFFFFF
+        size_val >>= 8
+
+    return f"{(~crc) & 0xFFFFFFFF:08x}"
+
+
+def _crc32_broken(data: bytes) -> str:
+    """
+    Similar to posix but with a broken length calculation.
+
+    Used in many BLUE files.
+    """
+    table = _generate_crc_table(0x04C11DB7)
+    crc = 0xFFFFFFFF
+
+    # process data bytes
+    for byte_val in data:
+        crc = (crc << 8) ^ table[(crc >> 24) ^ byte_val]
+        crc &= 0xFFFFFFFF
+
+    # broken length calculation - only processes high byte of length
+    size_val = len(data)
+    while size_val > 0:
+        crc = (crc << 8) ^ table[(crc >> 24) ^ (size_val >> 24)]
+        crc &= 0xFFFFFFFF
+        size_val >>= 8
+
+    return f"{(~crc) & 0xFFFFFFFF:08x}"
+
+
 def _get_blue_boundaries(blue_path: Path, h_fixed: dict) -> (int, int):
     """
     Extract data boundaries from fixed header.
@@ -395,7 +467,6 @@ def _description(h_fixed: dict) -> str:
             spec_str = f"Platinum {version}"
     except InvalidVersion:
         log.warning("Could not parse BLUE specification from VER keyword.")
-        pass
     # h_fixed will contain number e.g. 1000, 1001, 2000, 2001
     description = (
         f"Read {h_fixed['version']} type {h_fixed['type']} {h_fixed['format']} using {spec_str} specification."
@@ -460,7 +531,7 @@ def get_tag(tag):
     global_info = {
         SigMFFile.AUTHOR_KEY: getpass.getuser(),
         SigMFFile.DATATYPE_KEY: datatype,
-        SigMFFile.RECORDER_KEY: f"Official SigMF BLUE converter",
+        SigMFFile.RECORDER_KEY: "Official SigMF BLUE converter",
         SigMFFile.NUM_CHANNELS_KEY: num_channels,
         SigMFFile.SAMPLE_RATE_KEY: sample_rate_hz,
         SigMFFile.EXTENSIONS_KEY: [{"name": "blue", "version": "0.0.1", "optional": True}],