Improved USLP checksum handling

Author: robamu

Diff for: CHANGELOG.md

@@ -8,6 +8,16 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 # [unreleased]
 
+## Changed
+
+- USLP implementation: FECF is fixed to 2 bytes and always uses the standard CRC16 CCITT checksum
+  calculation now. Consequently, the USLP API was adapted and simplified.
+  - `FecfProperties` removed, not required anymore
+  - `TransferFrame` constructor now expects `has_fecf` boolean parameter which defaults to true.
+     The checksum will be calculated and appended by the `pack` method. The `unpack` method
+     expects the `has_fecf` flag now as well and will perform a CRC16 calculation when the flag
+     is set to True.
+
 # [v0.26.1] 2024-11-30
 
 ## Fixed

Diff for: src/spacepackets/ecss/pus_1_verification.py

@@ -51,7 +51,10 @@ def len(self) -> int:
 
     @classmethod
     def unpack(
-        cls, data: bytes | bytearray, num_bytes_err_code: int, num_bytes_data: int | None = None
+        cls,
+        data: bytes | bytearray,
+        num_bytes_err_code: int,
+        num_bytes_data: int | None = None,
     ) -> FailureNotice:
         pfc = num_bytes_err_code * 8
         if num_bytes_data is None:
@@ -281,7 +284,10 @@ def error_code(self) -> ErrorCode | None:
 
     @property
     def is_step_reply(self) -> bool:
-        return self.subservice in (Subservice.TM_STEP_FAILURE, Subservice.TM_STEP_SUCCESS)
+        return self.subservice in (
+            Subservice.TM_STEP_FAILURE,
+            Subservice.TM_STEP_SUCCESS,
+        )
 
     @property
     def step_id(self) -> StepId | None:

Diff for: src/spacepackets/uslp/defs.py

@@ -24,3 +24,7 @@ class UslpVersionMissmatchError(Exception):
 
 class UslpTypeMissmatchError(Exception):
     pass
+
+
+class UslpChecksumError(Exception):
+    pass

Diff for: src/spacepackets/uslp/frame.py

@@ -4,7 +4,10 @@
 import struct
 from typing import Union
 
+from spacepackets.crc import CRC16_CCITT_FUNC
+
 from .defs import (
+    UslpChecksumError,
     UslpFhpVhopFieldMissingError,
     UslpInvalidConstructionRulesError,
     UslpInvalidFrameHeaderError,
@@ -30,28 +33,19 @@ def __init__(self, present: bool, size: int):
         self.size = size
 
 
-class FecfProperties:
-    def __init__(self, present: bool, size: int):
-        self.present = present
-        self.size = size
-
-
 class FramePropertiesBase:
     def __init__(
         self,
         has_insert_zone: bool,
         has_fecf: bool,
         insert_zone_len: int | None = None,
-        fecf_len: int | None = None,
     ):
         if has_insert_zone and insert_zone_len is None:
             raise ValueError
-        if has_fecf and fecf_len is None:
-            raise ValueError
         self.insert_zone_properties = InsertZoneProperties(
             present=has_insert_zone, size=insert_zone_len
         )
-        self.fecf_properties = FecfProperties(present=has_fecf, size=fecf_len)
+        self.has_fecf = has_fecf
 
 
 class FixedFrameProperties(FramePropertiesBase):
@@ -61,7 +55,6 @@ def __init__(
         has_insert_zone: bool,
         has_fecf: bool,
         insert_zone_len: int | None = None,
-        fecf_len: int | None = None,
     ):
         """Contains properties required when unpacking fixed USLP frames. These properties
         can not be determined by parsing the frame. The standard refers to these properties
@@ -76,7 +69,6 @@ def __init__(
             has_insert_zone=has_insert_zone,
             has_fecf=has_fecf,
             insert_zone_len=insert_zone_len,
-            fecf_len=fecf_len,
         )
         self.fixed_len = fixed_len
 
@@ -88,7 +80,6 @@ def __init__(
         has_fecf: bool,
         truncated_frame_len: int,
         insert_zone_len: int | None = None,
-        fecf_len: int | None = None,
     ):
         """Contains properties required when unpacking variable USLP frames. These properties
         can not be determined by parsing the frame. The standard refers to these properties
@@ -104,7 +95,6 @@ def __init__(
             has_insert_zone=has_insert_zone,
             has_fecf=has_fecf,
             insert_zone_len=insert_zone_len,
-            fecf_len=fecf_len,
         )
         self.truncated_frame_len = truncated_frame_len
 
@@ -324,13 +314,13 @@ def __init__(
         tfdf: TransferFrameDataField,
         insert_zone: bytes | None = None,
         op_ctrl_field: bytes | None = None,
-        fecf: bytes | None = None,
+        has_fecf: bool = True,
     ):
         self.header = header
         self.tfdf = tfdf
         self.insert_zone = insert_zone
         self.op_ctrl_field = op_ctrl_field
-        self.fecf = fecf
+        self.has_fecf = has_fecf
 
     def pack(self, truncated: bool = False, frame_type: FrameType | None = None) -> bytearray:
         frame = bytearray()
@@ -346,8 +336,8 @@ def pack(self, truncated: bool = False, frame_type: FrameType | None = None) ->
             frame.extend(self.op_ctrl_field)
         elif not truncated and self.header.op_ctrl_flag:
             raise UslpInvalidFrameHeaderError
-        if self.fecf is not None:
-            frame.extend(self.fecf)
+        if self.has_fecf:
+            frame.extend(struct.pack("!H", CRC16_CCITT_FUNC(frame)))
         return frame
 
     def set_frame_len_in_header(self) -> None:
@@ -363,8 +353,8 @@ def len(self) -> int:
             size += len(self.insert_zone)
         if self.op_ctrl_field is not None:
             size += len(self.op_ctrl_field)
-        if self.fecf is not None:
-            size += len(self.fecf)
+        if self.has_fecf:
+            size += 2
         return size
 
     @classmethod
@@ -383,13 +373,16 @@ def __empty(cls) -> TransferFrame:
             tfdf=empty_data_field,
             insert_zone=None,
             op_ctrl_field=None,
-            fecf=None,
+            has_fecf=False,
         )
 
     # TODO: Fix lint by creating helper methods.
     @classmethod
     def unpack(  # noqa: PLR0912 too many branches
-        cls, raw_frame: bytes, frame_type: FrameType, frame_properties: FramePropertiesT
+        cls,
+        raw_frame: bytes | bytearray,
+        frame_type: FrameType,
+        frame_properties: FramePropertiesT,
     ) -> TransferFrame:
         """Unpack a USLP transfer frame from a raw bytearray. All managed parameters have
         to be passed explicitly.
@@ -457,11 +450,11 @@ def unpack(  # noqa: PLR0912 too many branches
             frame.op_ctrl_field = raw_frame[current_idx : current_idx + 4]
             current_idx += 4
         # Parse Frame Error Control field if present
-        if frame_properties.fecf_properties.present:
-            frame.fecf = raw_frame[
-                current_idx : current_idx + frame_properties.fecf_properties.size
-            ]
-            current_idx += frame_properties.fecf_properties.size
+        if frame_properties.has_fecf:
+            crc_check = CRC16_CCITT_FUNC(raw_frame[0 : current_idx + 2])
+            if crc_check != 0:
+                raise UslpChecksumError
+
         return frame
 
     @staticmethod
@@ -497,8 +490,8 @@ def __get_tfdf_len(
             exact_tfdf_len = properties.truncated_frame_len - header_len
         else:
             exact_tfdf_len = header.frame_len + 1 - header_len
-        if properties.fecf_properties.present:
-            exact_tfdf_len -= properties.fecf_properties.size
+        if properties.has_fecf:
+            exact_tfdf_len -= 2
         if header_type != HeaderType.TRUNCATED and header.op_ctrl_flag:
             exact_tfdf_len -= 4
         if properties.insert_zone_properties.present:

Diff for: src/spacepackets/uslp/header.py

@@ -183,9 +183,9 @@ def __init__(
         vcid: int,
         map_id: int,
         frame_len: int,
-        bypass_seq_ctrl_flag: BypassSequenceControlFlag,
-        prot_ctrl_cmd_flag: ProtocolCommandFlag,
-        op_ctrl_flag: bool,
+        bypass_seq_ctrl_flag: BypassSequenceControlFlag = BypassSequenceControlFlag.SEQ_CTRLD_QOS,
+        prot_ctrl_cmd_flag: ProtocolCommandFlag = ProtocolCommandFlag.USER_DATA,
+        op_ctrl_flag: bool = False,
         vcf_count_len: int = 0,
         vcf_count: int | None = None,
     ):
@@ -286,7 +286,7 @@ def len(self) -> int:
         return 7 + self.vcf_count_len
 
 
-def determine_header_type(header_start: bytes) -> HeaderType:
+def determine_header_type(header_start: bytes | bytearray) -> HeaderType:
     """Determine header type from raw header.
     :param header_start:
     :raises ValueError: Passed bytearray shorter than minimum length 4

Diff for: tests/cfdp/tlvslvs/test_msg_to_user.py

@@ -1,6 +1,11 @@
 from unittest import TestCase
 
-from spacepackets.cfdp import MessageToUserTlv, TlvHolder, TlvType, TlvTypeMissmatchError
+from spacepackets.cfdp import (
+    MessageToUserTlv,
+    TlvHolder,
+    TlvType,
+    TlvTypeMissmatchError,
+)
 from spacepackets.cfdp.tlv import (
     CfdpTlv,
     ProxyMessageType,