diff --git a/docs/changelog.md b/docs/changelog.md index df2d4ce8e..288585e61 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -8,6 +8,20 @@ nav_order: 2 # Unreleased changes +### Breaking changes + +- `CEMILData.flags` is now a `CEMIFlags` object instead of a 16 bit `int`. `CEMIFlags` was previously a collection of bit constants; it is now a slotted dataclass holding the control field values that are independent state: + - `priority: CEMIPriority` - new `SYSTEM` / `NORMAL` / `URGENT` / `LOW` enum + - `repeat_on_error: bool`, `system_broadcast: bool` - named for the positive meaning; both are inverted on the wire + - `acknowledge_request: bool` + - `confirm_error: bool` - only meaningful in an L_Data.con frame + - `hop_count: int` - replaces the `CEMILData.hops` property, which was removed + - `frame_type: CEMIFrameType` and `frame_format: CEMIFrameFormat` - informational; they hold what was received and are ignored when serializing + The Frame Type of an outgoing frame is derived from its NPDU length and the Address Type from the type of the destination address (`CEMILData.address_type`), so neither can disagree with the frame that is put on the wire. `CEMIFrameType` and the new `CEMIAddressType` place their own bit via `to_knx()` / `from_knx()`; `CEMILData` composes the control field as `flags.to_knx() | frame_type.to_knx() | address_type.to_knx()`. Ctrl1 and Ctrl2 are handled as one 16 bit int, so `CEMIFlags.to_knx()` returns an `int` and `CEMIFlags.from_knx()` takes one. The remaining bit masks moved to `xknx.cemi.flags` as module level constants. +- `CEMIFlags` has a compact `__str__` used in `CEMILData.__repr__` - eg. `LOW STANDARD hop_count=6`, listing only the boolean flags that are set. The full dataclass `__repr__` shows every field. +- `CEMILData(flags=...)` is optional now and defaults to `CEMIFlags()` - low priority, hop count 6, no acknowledge request. +- `xknx.secure.data_secure_asdu.block_0()`, `SecureData.init_from_plain_apdu()` and `SecureData.get_plain_apdu()` take `address_type: CEMIAddressType` instead of `frame_flags: int`. Only the Address Type bit of Ctrl2 ever reached the CCM input, so the B0 flags octet is `address_type.to_knx()` and `B0_AT_FIELD_FLAGS_MASK` is gone. + ### Bugfixes - Send frames with an APDU longer than 15 octets as L_Data_Extended frames. The Frame Type flag of Ctrl1 is now derived from the NPDU length when serializing a `CEMILData` instead of always being set to standard frame. This fixes sending telegrams over KNX Data Secure with a payload of 2 octets or more (eg. DPT 9.x or DPT 232.600) - Data Secure adds 12 octets to the plain APDU, which no longer fits in a standard frame. Sending an APDU longer than 254 octets now raises `ConversionError` instead of `OverflowError`. @@ -18,8 +32,9 @@ nav_order: 2 ### Protocol -- Reject incoming CEMI L_Data frames with a non-zero Extended Frame Format field with `UnsupportedCEMIMessage`. LTE-HEE frames use zone addressing, so their address fields can not be parsed as `GroupAddress`. The Frame Type flag is still not validated against the NPDU length when parsing - a receiver shall be tolerant towards the used frame format. -- `CEMIFlags.EXTENDED_FRAME_FORMAT` was removed; its value `0x0001` was reserved, not an "extended frame format" indicator - `0x0000` is used for standard frames as well as for long extended frames. `CEMIFlags.LTE_FRAME_FORMAT` and `CEMIFlags.EXTENDED_FRAME_FORMAT_MASK` were added instead. +- Reject incoming CEMI L_Data frames with a non-standard Extended Frame Format field with `UnsupportedCEMIMessage`. LTE-HEE frames use zone addressing, so their address fields can not be parsed as `GroupAddress`; reserved encodings are rejected too. The Frame Type flag is still not validated against the NPDU length when parsing - a receiver shall be tolerant towards the used frame format. +- `CEMIFlags.EXTENDED_FRAME_FORMAT` was removed; its value `0x0001` was reserved, not an "extended frame format" indicator - `0x0000` is used for standard frames as well as for long extended frames. The `CEMIFrameFormat` enum covers the field instead, resolving the whole `01xxb` range to `LTE_HEE`. +- The hop count is validated when serializing; a value outside `0..7` raises `ConversionError` instead of silently corrupting Ctrl2. - Add explicit length checks to every remaining APCI `from_knx` (and the top-level `APCI.from_knx` dispatcher) as defense-in-depth on top of the broad `except (IndexError, struct.error, ValueError)` added in 3.17.0: each service now raises `ConversionError` with a specific "Invalid length for A_X in CEMI" message for a truncated, malformed or overlong frame instead of relying solely on the generic dispatcher-level catch. # 3.17.0 APCIs and DPTs 2026-07-25 diff --git a/test/cemi_tests/cemi_frame_test.py b/test/cemi_tests/cemi_frame_test.py index 6a8c56663..7424f8ba2 100644 --- a/test/cemi_tests/cemi_frame_test.py +++ b/test/cemi_tests/cemi_frame_test.py @@ -3,14 +3,17 @@ import pytest from xknx.cemi import ( + CEMIAddressType, CEMIFlags, CEMIFrame, + CEMIFrameType, CEMILData, CEMIMessageCode, CEMIMPropReadRequest, CEMIMPropReadResponse, CEMIMPropWriteRequest, CEMIMPropWriteResponse, + CEMIPriority, ) from xknx.cemi.const import CEMIErrorCode from xknx.dpt import DPTArray @@ -56,8 +59,12 @@ def test_valid_command() -> None: frame = CEMIFrame.from_knx(raw) assert frame.code == CEMIMessageCode.L_DATA_IND assert isinstance(frame.data, CEMILData) - assert frame.data.flags == 0x8080 - assert frame.data.hops == 0 + assert frame.data.flags == CEMIFlags( + priority=CEMIPriority.SYSTEM, + repeat_on_error=True, + system_broadcast=True, + hop_count=0, + ) assert frame.data.src_addr == IndividualAddress(1) assert frame.data.dst_addr == GroupAddress(1) assert frame.data.payload == GroupValueRead() @@ -72,8 +79,12 @@ def test_valid_tpci_control() -> None: frame = CEMIFrame.from_knx(raw) assert frame.code == CEMIMessageCode.L_DATA_IND assert isinstance(frame.data, CEMILData) - assert frame.data.flags == 0x8000 - assert frame.data.hops == 0 + assert frame.data.flags == CEMIFlags( + priority=CEMIPriority.SYSTEM, + repeat_on_error=True, + system_broadcast=True, + hop_count=0, + ) assert frame.data.payload is None assert frame.data.src_addr == IndividualAddress(0) assert frame.data.dst_addr == IndividualAddress(0) @@ -160,7 +171,6 @@ def test_invalid_payload() -> None: frame = CEMIFrame( code=CEMIMessageCode.L_DATA_IND, data=CEMILData( - flags=0, src_addr=IndividualAddress(0), dst_addr=IndividualAddress(0), tpci=TDataGroup(), @@ -235,8 +245,8 @@ def test_telegram_group_address() -> None: data=CEMILData.init_from_telegram(_telegram), ) assert isinstance(frame.data, CEMILData) - assert frame.data.flags & 0x0080 == CEMIFlags.DESTINATION_GROUP_ADDRESS - assert frame.data.flags & 0x0C00 == CEMIFlags.PRIORITY_LOW + assert frame.data.address_type is CEMIAddressType.GROUP + assert frame.data.flags == CEMIFlags(priority=CEMIPriority.LOW) # test CEMIFrame.telegram property assert frame.data.telegram() == _telegram @@ -249,8 +259,8 @@ def test_telegram_broadcast() -> None: data=CEMILData.init_from_telegram(_telegram), ) assert isinstance(frame.data, CEMILData) - assert frame.data.flags & 0x0080 == CEMIFlags.DESTINATION_GROUP_ADDRESS - assert frame.data.flags & 0x0C00 == CEMIFlags.PRIORITY_SYSTEM + assert frame.data.address_type is CEMIAddressType.GROUP + assert frame.data.flags == CEMIFlags(priority=CEMIPriority.SYSTEM) assert frame.data.tpci == TDataBroadcast() # test CEMIFrame.telegram property assert frame.data.telegram() == _telegram @@ -264,9 +274,9 @@ def test_telegram_individual_address() -> None: data=CEMILData.init_from_telegram(_telegram), ) assert isinstance(frame.data, CEMILData) - assert frame.data.flags & 0x0080 == CEMIFlags.DESTINATION_INDIVIDUAL_ADDRESS - assert frame.data.flags & 0x0C00 == CEMIFlags.PRIORITY_SYSTEM - assert frame.data.flags & 0x0200 == CEMIFlags.NO_ACK_REQUESTED + assert frame.data.address_type is CEMIAddressType.INDIVIDUAL + assert frame.data.flags == CEMIFlags(priority=CEMIPriority.SYSTEM) + assert not frame.data.flags.acknowledge_request # test CEMIFrame.telegram property assert frame.data.telegram() == _telegram @@ -289,45 +299,41 @@ def _cemi_l_data_from_payload(payload: GroupValueWrite) -> bytes: @pytest.mark.parametrize( - "apdu_payload_length,expected_npdu_len,expected_frame_type", + "apdu_payload_length,expected_npdu_len,expected_standard_frame", [ - (1, 2, CEMIFlags.FRAME_TYPE_STANDARD), + (1, 2, True), # 15 octets after the TPCI octet is the maximum of a standard frame - (14, 15, CEMIFlags.FRAME_TYPE_STANDARD), - (15, 16, CEMIFlags.FRAME_TYPE_EXTENDED), - (253, 254, CEMIFlags.FRAME_TYPE_EXTENDED), + (14, 15, True), + (15, 16, False), + (253, 254, False), ], ) def test_frame_type_from_npdu_length( - apdu_payload_length: int, expected_npdu_len: int, expected_frame_type: int + apdu_payload_length: int, expected_npdu_len: int, expected_standard_frame: bool ) -> None: """Test Frame Type flag is derived from the NPDU length.""" raw = _cemi_l_data_from_payload( GroupValueWrite(DPTArray(bytes(apdu_payload_length))) ) assert raw[6] == expected_npdu_len - assert (raw[0] << 8) & CEMIFlags.FRAME_TYPE_STANDARD == expected_frame_type + assert ( + CEMIFrameType.from_knx(raw[0] << 8) is CEMIFrameType.STANDARD + ) is expected_standard_frame -def test_frame_type_overrides_flags() -> None: - """Test Frame Type flag of `flags` is overridden by the payload length.""" - long_payload = GroupValueWrite(DPTArray(bytes(15))) - short_payload = GroupValueWrite(DPTArray(bytes(1))) +def test_frame_type_follows_payload() -> None: + """Test the Frame Type follows the current payload; it is not held by `flags`.""" cemi_data = CEMILData( - # standard frame flag set although the payload requires an extended frame - flags=CEMIFlags.FRAME_TYPE_STANDARD | CEMIFlags.DESTINATION_GROUP_ADDRESS, src_addr=IndividualAddress(1), dst_addr=GroupAddress(1), tpci=TDataGroup(), - payload=long_payload, + payload=GroupValueWrite(DPTArray(bytes(15))), ) - assert not cemi_data.to_knx()[0] & 0x80 - # `flags` is not modified by serialization - assert cemi_data.flags & CEMIFlags.FRAME_TYPE_STANDARD + assert CEMIFrameType.from_knx(cemi_data.to_knx()[0] << 8) is CEMIFrameType.EXTENDED - cemi_data.flags = CEMIFlags.DESTINATION_GROUP_ADDRESS # extended frame flag - cemi_data.payload = short_payload - assert cemi_data.to_knx()[0] & 0x80 + # replacing the payload - as Data Secure does - changes the Frame Type + cemi_data.payload = GroupValueWrite(DPTArray(bytes(1))) + assert CEMIFrameType.from_knx(cemi_data.to_knx()[0] << 8) is CEMIFrameType.STANDARD def test_npdu_length_exceeded() -> None: @@ -336,23 +342,30 @@ def test_npdu_length_exceeded() -> None: _cemi_l_data_from_payload(GroupValueWrite(DPTArray(bytes(254)))) -def test_extended_frame_format_not_supported() -> None: +@pytest.mark.parametrize( + "eff,err_msg", + [ + # 01xxb - LTE-HEE zone addressed frames + (0b0100, r".*Extended Frame Format not supported: LTE_HEE.*"), + (0b0111, r".*Extended Frame Format not supported: LTE_HEE.*"), + # reserved + (0b0001, r".*Reserved Extended Frame Format: 0b0001.*"), + (0b1111, r".*Reserved Extended Frame Format: 0b1111.*"), + ], +) +def test_extended_frame_format_not_supported(eff: int, err_msg: str) -> None: """Test parsing of LTE-HEE and reserved Extended Frame Formats.""" raw = get_data( 0x29, 0, - CEMIFlags.FRAME_TYPE_EXTENDED - | CEMIFlags.DESTINATION_GROUP_ADDRESS - | CEMIFlags.LTE_FRAME_FORMAT, + (CEMIAddressType.GROUP.to_knx() | eff), # Ctrl2; Ctrl1 = extended frame 1, 1, 1, 0, [], ) - with pytest.raises( - UnsupportedCEMIMessage, match=r".*Extended Frame Format not supported.*" - ): + with pytest.raises(UnsupportedCEMIMessage, match=err_msg): CEMIFrame.from_knx(raw) diff --git a/test/cemi_tests/flags_test.py b/test/cemi_tests/flags_test.py new file mode 100644 index 000000000..7ed4aeb84 --- /dev/null +++ b/test/cemi_tests/flags_test.py @@ -0,0 +1,172 @@ +"""Tests for the cEMI L_Data control fields.""" + +import pytest + +from xknx.cemi import ( + CEMIAddressType, + CEMIFlags, + CEMIFrameFormat, + CEMIFrameType, + CEMIPriority, +) +from xknx.exceptions import ConversionError + + +def control_field(ctrl1: int, ctrl2: int) -> int: + """Return Ctrl1 and Ctrl2 as one 16 bit control field.""" + return ctrl1 << 8 | ctrl2 + + +@pytest.mark.parametrize( + "ctrl1,ctrl2,flags", + [ + ( + # Ctrl1 0xBC: standard frame, do not repeat, broadcast, low priority + # Ctrl2 0xE0: group address, hop count 6, standard frame format + 0xBC, + 0xE0, + CEMIFlags(priority=CEMIPriority.LOW, hop_count=6), + ), + ( + # Ctrl1 0x3C: extended frame - kept, but not evaluated + 0x3C, + 0xE0, + CEMIFlags( + priority=CEMIPriority.LOW, + hop_count=6, + frame_type=CEMIFrameType.EXTENDED, + ), + ), + ( + # Ctrl1 0xB0: system priority; Ctrl2 0x60: individual address + 0xB0, + 0x60, + CEMIFlags(priority=CEMIPriority.SYSTEM, hop_count=6), + ), + ( + # every optional flag set: repeat on error, system broadcast, + # acknowledge requested, confirm error, urgent priority, hop count 7 + 0b1000_1011, + 0b0111_0000, + CEMIFlags( + priority=CEMIPriority.URGENT, + repeat_on_error=True, + system_broadcast=True, + acknowledge_request=True, + confirm_error=True, + hop_count=7, + ), + ), + ], +) +def test_from_knx(ctrl1: int, ctrl2: int, flags: CEMIFlags) -> None: + """Test parsing of the control field.""" + assert CEMIFlags.from_knx(control_field(ctrl1, ctrl2)) == flags + + +@pytest.mark.parametrize( + "eff,frame_format", + [ + (0b0000, CEMIFrameFormat.STANDARD), + # the whole 01xxb range is LTE-HEE; the 2 lsb are the LTE address type + (0b0100, CEMIFrameFormat.LTE_HEE), + (0b0101, CEMIFrameFormat.LTE_HEE), + (0b0110, CEMIFrameFormat.LTE_HEE), + (0b0111, CEMIFrameFormat.LTE_HEE), + ], +) +def test_frame_format_from_knx(eff: int, frame_format: CEMIFrameFormat) -> None: + """Test parsing of the Extended Frame Format field.""" + flags = CEMIFlags.from_knx(control_field(0xBC, 0xE0 | eff)) + assert flags.frame_format is frame_format + + +@pytest.mark.parametrize("eff", [0b0001, 0b0010, 0b0011, 0b1000, 0b1100, 0b1111]) +def test_reserved_frame_format(eff: int) -> None: + """Test parsing of a reserved Extended Frame Format.""" + with pytest.raises(ConversionError, match=r".*Reserved Extended Frame Format.*"): + CEMIFlags.from_knx(control_field(0xBC, 0xE0 | eff)) + + +@pytest.mark.parametrize( + "priority,ctrl1", + [ + (CEMIPriority.SYSTEM, 0b0011_0000), + (CEMIPriority.NORMAL, 0b0011_0100), + (CEMIPriority.URGENT, 0b0011_1000), + (CEMIPriority.LOW, 0b0011_1100), + ], +) +def test_priority_to_knx(priority: CEMIPriority, ctrl1: int) -> None: + """Test priority encoding - 3/2/2 §2.2.2 Figure 28.""" + # the Frame Type bit is not set by `CEMIFlags.to_knx()` + assert CEMIFlags(priority=priority).to_knx() >> 8 == ctrl1 + + +def test_to_knx_leaves_derived_bits_clear() -> None: + """Test Frame Type and Address Type are not set by `CEMIFlags.to_knx()`.""" + raw = CEMIFlags().to_knx() + assert raw == control_field(0b0011_1100, 0b0110_0000) + assert CEMIFrameType.from_knx(raw) is CEMIFrameType.EXTENDED # ie. bit not set + assert CEMIAddressType.from_knx(raw) is CEMIAddressType.INDIVIDUAL + + +@pytest.mark.parametrize( + "frame_type,address_type,expected", + [ + (CEMIFrameType.STANDARD, CEMIAddressType.GROUP, (0xBC, 0xE0)), + (CEMIFrameType.EXTENDED, CEMIAddressType.GROUP, (0x3C, 0xE0)), + (CEMIFrameType.STANDARD, CEMIAddressType.INDIVIDUAL, (0xBC, 0x60)), + (CEMIFrameType.EXTENDED, CEMIAddressType.INDIVIDUAL, (0x3C, 0x60)), + ], +) +def test_composition( + frame_type: CEMIFrameType, + address_type: CEMIAddressType, + expected: tuple[int, int], +) -> None: + """Test the frame ORs the derived bits into the flags.""" + raw = CEMIFlags().to_knx() | frame_type.to_knx() | address_type.to_knx() + assert raw == control_field(*expected) + + +def test_to_knx_ignores_received_frame_type() -> None: + """Test the Frame Type of a received frame is not used when serializing.""" + flags = CEMIFlags.from_knx(control_field(0x3C, 0xE0)) + assert flags.frame_type is CEMIFrameType.EXTENDED + raw = ( + flags.to_knx() + | CEMIFrameType.STANDARD.to_knx() + | CEMIAddressType.GROUP.to_knx() + ) + assert raw == control_field(0xBC, 0xE0) + + +def test_round_trip() -> None: + """Test parsing and serializing yields the same control field.""" + for ctrl1 in (0xBC, 0xB0, 0x9C, 0xBF, 0x3C): + for ctrl2 in (0xE0, 0x60, 0xF0, 0x00): + raw = control_field(ctrl1, ctrl2) + flags = CEMIFlags.from_knx(raw) + assert ( + flags.to_knx() + | flags.frame_type.to_knx() + | CEMIAddressType.from_knx(raw).to_knx() + ) == raw + + +@pytest.mark.parametrize("hop_count", [-1, 8, 255]) +def test_invalid_hop_count(hop_count: int) -> None: + """Test hop count out of range.""" + with pytest.raises(ConversionError, match=r".*Hop count out of range.*"): + CEMIFlags(hop_count=hop_count).to_knx() + + +def test_str() -> None: + """Test the compact string representation only lists flags that are set.""" + assert str(CEMIFlags()) == "LOW STANDARD hop_count=6" + assert ( + str(CEMIFlags.from_knx(control_field(0b1000_1011, 0b0111_0000))) + == "URGENT STANDARD hop_count=7 repeat_on_error system_broadcast " + "acknowledge_request confirm_error" + ) diff --git a/test/knxip_tests/routing_indication_test.py b/test/knxip_tests/routing_indication_test.py index 558720fde..53daf3e57 100644 --- a/test/knxip_tests/routing_indication_test.py +++ b/test/knxip_tests/routing_indication_test.py @@ -44,7 +44,7 @@ def test_telegram_set(self) -> None: ), ) assert isinstance(cemi.data, CEMILData) - cemi.data.hops = 5 + cemi.data.flags.hop_count = 5 routing_indication = RoutingIndication(raw_cemi=cemi.to_knx()) knxipframe = KNXIPFrame.init_from_body(routing_indication) diff --git a/test/secure_tests/data_secure_test.py b/test/secure_tests/data_secure_test.py index f781b0ada..b13d5b191 100644 --- a/test/secure_tests/data_secure_test.py +++ b/test/secure_tests/data_secure_test.py @@ -7,7 +7,7 @@ import pytest from xknx import XKNX -from xknx.cemi import CEMIFlags, CEMIFrame, CEMILData, CEMIMessageCode +from xknx.cemi import CEMIFrame, CEMIFrameType, CEMILData, CEMIMessageCode from xknx.dpt import DPTArray from xknx.exceptions import DataSecureError from xknx.secure.data_secure import is_data_secure @@ -482,7 +482,7 @@ def test_data_secure_authentication_only(self) -> None: # 4 octet plain APDU + 12 octets Data Secure overhead doesn't fit in a # standard frame - it is serialized as L_Data_Extended frame assert outgoing_raw[6] == 16 - assert not outgoing_raw[0] & CEMIFlags.FRAME_TYPE_STANDARD >> 8 + assert CEMIFrameType.from_knx(outgoing_raw[0] << 8) is CEMIFrameType.EXTENDED # create new cemi to avoid mixed bytearray / byte parts incoming_cemi = CEMIFrame.from_knx(b"\x11\x00" + outgoing_raw) diff --git a/test/str_test.py b/test/str_test.py index 2536775dd..c48fcb990 100644 --- a/test/str_test.py +++ b/test/str_test.py @@ -735,7 +735,8 @@ def test_cemi_ldata_frame(self) -> None: assert ( str(cemi_frame) == '" />")" />' ) diff --git a/xknx/cemi/__init__.py b/xknx/cemi/__init__.py index 1d33c1b8c..3e6d76b23 100644 --- a/xknx/cemi/__init__.py +++ b/xknx/cemi/__init__.py @@ -11,4 +11,11 @@ CEMIMPropWriteResponse, ) from .cemi_handler import CEMIHandler -from .const import CEMIErrorCode, CEMIFlags, CEMIMessageCode +from .const import CEMIErrorCode, CEMIMessageCode +from .flags import ( + CEMIAddressType, + CEMIFlags, + CEMIFrameFormat, + CEMIFrameType, + CEMIPriority, +) diff --git a/xknx/cemi/cemi_frame.py b/xknx/cemi/cemi_frame.py index 113f64539..ca725af37 100644 --- a/xknx/cemi/cemi_frame.py +++ b/xknx/cemi/cemi_frame.py @@ -31,9 +31,15 @@ MAX_NPDU_LENGTH, STANDARD_FRAME_MAX_NPDU_LENGTH, CEMIErrorCode, - CEMIFlags, CEMIMessageCode, ) +from .flags import ( + CEMIAddressType, + CEMIFlags, + CEMIFrameFormat, + CEMIFrameType, + CEMIPriority, +) class CEMIInfo: @@ -103,31 +109,27 @@ class CEMILData(CEMIData): def __init__( self, *, - flags: int, + flags: CEMIFlags | None = None, src_addr: IndividualAddress, dst_addr: GroupAddress | IndividualAddress, tpci: TPCI, payload: APCI | None, ) -> None: """Initialize CEMILData object.""" - self.flags = flags + self.flags = flags if flags is not None else CEMIFlags() self.src_addr = src_addr self.dst_addr = dst_addr self.tpci = tpci self.payload = payload @property - def hops(self) -> int: - """Return hops.""" - return (self.flags & 0x0070) >> 4 - - @hops.setter - def hops(self, val: int) -> None: - """Set hops.""" - # Resetting hops - self.flags &= 0xFFFF ^ 0x0070 - # Setting new hops - self.flags |= val << 4 + def address_type(self) -> CEMIAddressType: + """Return the Address Type of the destination address.""" + return ( + CEMIAddressType.GROUP + if isinstance(self.dst_addr, GroupAddress) + else CEMIAddressType.INDIVIDUAL + ) def calculated_length(self) -> int: """Get length of KNX/IP body.""" @@ -143,30 +145,19 @@ def init_from_telegram( src_addr: IndividualAddress | None = None, ) -> CEMILData: """Return CEMILData from a Telegram.""" - flags = ( - # default; the Frame Type is derived from the NPDU length in `to_knx()` - CEMIFlags.FRAME_TYPE_STANDARD - | CEMIFlags.DO_NOT_REPEAT - | CEMIFlags.BROADCAST - | CEMIFlags.NO_ACK_REQUESTED - | CEMIFlags.CONFIRM_NO_ERROR - | CEMIFlags.HOP_COUNT_1ST - ) if isinstance(telegram.destination_address, GroupAddress): - flags |= CEMIFlags.DESTINATION_GROUP_ADDRESS - if isinstance(telegram.tpci, TDataBroadcast): - flags |= CEMIFlags.PRIORITY_SYSTEM - else: - flags |= CEMIFlags.PRIORITY_LOW - elif isinstance(telegram.destination_address, IndividualAddress): - flags |= ( - CEMIFlags.DESTINATION_INDIVIDUAL_ADDRESS | CEMIFlags.PRIORITY_SYSTEM + priority = ( + CEMIPriority.SYSTEM + if isinstance(telegram.tpci, TDataBroadcast) + else CEMIPriority.LOW ) + elif isinstance(telegram.destination_address, IndividualAddress): + priority = CEMIPriority.SYSTEM else: raise TypeError() return CEMILData( - flags=flags, + flags=CEMIFlags(priority=priority), src_addr=src_addr or telegram.source_address, dst_addr=telegram.destination_address, tpci=telegram.tpci, @@ -208,14 +199,15 @@ def to_knx(self) -> bytes: # frame shall not be used when a standard frame suffices - 3/2/2 §2.2.5.1. # Derived here instead of when the frame is created because the payload may # be replaced afterwards - eg. wrapped in a SecureAPDU by Data Secure. - flags = ( - self.flags | CEMIFlags.FRAME_TYPE_STANDARD + frame_type = ( + CEMIFrameType.STANDARD if npdu_len <= STANDARD_FRAME_MAX_NPDU_LENGTH - else self.flags & ~CEMIFlags.FRAME_TYPE_STANDARD + else CEMIFrameType.EXTENDED ) - return ( - flags.to_bytes(2, "big") + ( + self.flags.to_knx() | frame_type.to_knx() | self.address_type.to_knx() + ).to_bytes(2, "big") + self.src_addr.to_knx() + self.dst_addr.to_knx() + npdu_len.to_bytes(1, "big") @@ -230,24 +222,32 @@ def from_knx(cls, raw: bytes) -> CEMILData: f"CEMI too small. Length: {len(raw)}; CEMI: {raw.hex()}" ) - # Control field 1 and Control field 2 - first 2 octets - flags = int.from_bytes(raw[0:2], "big") - src_addr = IndividualAddress.from_knx(raw[2:4]) + # Control field 1 and Control field 2 - first 2 octets + _control_field = int.from_bytes(raw[0:2], "big") + try: + flags = CEMIFlags.from_knx(_control_field) + except ConversionError as err: + raise UnsupportedCEMIMessage( + f"{err} from {src_addr} in CEMI: {raw.hex()}" + ) from err + # The Extended Frame Format defines how the address fields are to be - # interpreted - 3/2/2 §2.2.5.3. Only 0000b (standard and long frames) is - # supported; 01xxb are LTE-HEE zone addressed frames, the rest is reserved. - # The Frame Type flag itself is deliberately not validated against the NPDU - # length: "any receiver shall be tolerant towards the use of the Frame - # Format" - 3/6/3 §4.1.5.2.3. - if _eff := flags & CEMIFlags.EXTENDED_FRAME_FORMAT_MASK: + # interpreted - 3/2/2 §2.2.5.3. Only STANDARD (standard and long frames) is + # supported; LTE-HEE frames are zone addressed. The Frame Type is kept as + # received but deliberately not validated against the NPDU length: "any + # receiver shall be tolerant towards the use of the Frame Format" + # - 3/6/3 §4.1.5.2.3. + if flags.frame_format is not CEMIFrameFormat.STANDARD: raise UnsupportedCEMIMessage( - f"Extended Frame Format not supported: {_eff:#06b} " + f"Extended Frame Format not supported: {flags.frame_format.name} " f"from {src_addr} in CEMI: {raw.hex()}" ) - _dst_is_group_address = bool(flags & CEMIFlags.DESTINATION_GROUP_ADDRESS) + _dst_is_group_address = ( + CEMIAddressType.from_knx(_control_field) is CEMIAddressType.GROUP + ) dst_addr: GroupAddress | IndividualAddress = ( GroupAddress.from_knx(raw[4:6]) if _dst_is_group_address @@ -324,7 +324,7 @@ def __repr__(self) -> str: "CEMILData(" f'src_addr="{self.src_addr.__repr__()}" ' f'dst_addr="{self.dst_addr.__repr__()}" ' - f'flags="{self.flags:16b}" ' + f'flags="{self.flags}" ' f'tpci="{self.tpci}" ' f'payload="{self.payload}")' ) diff --git a/xknx/cemi/const.py b/xknx/cemi/const.py index 7eac1ac39..d6182cb94 100644 --- a/xknx/cemi/const.py +++ b/xknx/cemi/const.py @@ -47,57 +47,6 @@ class CEMIMessageCode(Enum): M_RESET_IND = 0xF0 # Reset confirmation -class CEMIFlags: - """Enum class for CEMI Flags.""" - - # Bit 1/7 - Frame Type (FT); see 3/6/3 EMI_IMI §4.1.4.3.2 - # Derived from the NPDU length when serializing - see `CEMILData.to_knx`. - FRAME_TYPE_EXTENDED = 0x0000 - FRAME_TYPE_STANDARD = 0x8000 - - # Bit 1/6 - Reserved - - # Bit 1/5 - # Repeat in case of an error - REPEAT = 0x0000 - DO_NOT_REPEAT = 0x2000 - - # Bit 1/4 - SYSTEM_BROADCAST = 0x0000 - BROADCAST = 0x1000 - - # Bit 1/3+2 - PRIORITY_SYSTEM = 0x0000 - PRIORITY_NORMAL = 0x0400 - PRIORITY_URGENT = 0x0800 - PRIORITY_LOW = 0x0C00 - - # Bit 1/1 - NO_ACK_REQUESTED = 0x0000 - ACK_REQUESTED = 0x0200 - - # Bit 1/0 - CONFIRM_NO_ERROR = 0x0000 - CONFIRM_ERROR = 0x0100 - - # Bit 0/7 - DESTINATION_INDIVIDUAL_ADDRESS = 0x0000 - DESTINATION_GROUP_ADDRESS = 0x0080 - - # Bit 0/6+5+4 - HOP_COUNT_NO = 0x0070 - HOP_COUNT_1ST = 0x0060 - - # Bit 0/3+2+1+0 - Extended Frame Format (EFF); see 3/6/3 EMI_IMI §4.1.4.3.2 - # and 3/2/2 Communication Medium TP1 §2.2.5.3. - # 0000b is used for L_Data_Standard frames as well as for long - # L_Data_Extended frames; 01xxb denotes LTE-HEE (zone addressed) frames. - # All other values are reserved. - STANDARD_FRAME_FORMAT = 0x0000 - LTE_FRAME_FORMAT = 0x0004 - EXTENDED_FRAME_FORMAT_MASK = 0x000F - - class CEMIErrorCode(IntEnum): """Enum class for CEMI Error Codes.""" diff --git a/xknx/cemi/flags.py b/xknx/cemi/flags.py new file mode 100644 index 000000000..4e465e35d --- /dev/null +++ b/xknx/cemi/flags.py @@ -0,0 +1,207 @@ +""" +Control fields of a cEMI L_Data frame. + +The two control field octets Ctrl1 and Ctrl2 are specified in +3/6/3 EMI_IMI §4.1.4.3.2 and 3/2/2 Communication Medium TP1 §2.2.2 / §2.2.5.3. + + Ctrl1 Ctrl2 + 7 6 5 4 3 2 1 0 7 6 5 4 3 2 1 0 + FT r R SB P P A C AT H H H E E E E + +Not every field is independent state: the Frame Type (FT) follows from the NPDU +length and the Address Type (AT) from the destination address, so both are +derived when serializing - see `CEMILData`. The received Frame Type is still kept +for inspection; the Address Type can be read from the type of the destination +address. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import IntEnum + +from xknx.exceptions import ConversionError + +# Ctrl1 and Ctrl2 are handled as one 16 bit value: Ctrl1 << 8 | Ctrl2. +# The Frame Type and Address Type bits are not listed here - `CEMIFrameType` and +# `CEMIAddressType` place their own bit. +# Ctrl1 +DO_NOT_REPEAT = 0b00100000_00000000 +BROADCAST = 0b00010000_00000000 +PRIORITY_MASK = 0b00001100_00000000 +PRIORITY_OFFSET = 10 +ACK_REQUESTED = 0b00000010_00000000 +CONFIRM_ERROR = 0b00000001_00000000 + +# Ctrl2 +HOP_COUNT_MASK = 0b00000000_01110000 +HOP_COUNT_OFFSET = 4 +EXTENDED_FRAME_FORMAT_MASK = 0b00000000_00001111 + +MAX_HOP_COUNT = 7 + + +class CEMIPriority(IntEnum): + """Priority of a cEMI L_Data frame. See 3/2/2 §2.2.2 Figure 28.""" + + SYSTEM = 0b00 + NORMAL = 0b01 + URGENT = 0b10 + LOW = 0b11 + + +class CEMIFrameType(IntEnum): + """ + Frame Type of a cEMI L_Data frame. See 3/6/3 §4.1.4.3.2. + + An L_Data_Standard frame carries at most 15 octets after the TPCI octet; + longer APDUs require an L_Data_Extended frame - 3/2/2 §2.2.4 and §2.2.5.1. + """ + + EXTENDED = 0 + STANDARD = 1 + + def to_knx(self) -> int: + """Serialize to the Frame Type bit of the control field - Ctrl1 b7.""" + return self << 15 + + @classmethod + def from_knx(cls, raw: int) -> CEMIFrameType: + """Parse the Frame Type bit from the control field.""" + return cls(raw >> 15 & 0b1) + + +class CEMIAddressType(IntEnum): + """ + Destination Address Type of a cEMI L_Data frame. See 3/6/3 §4.1.4.3.2. + + Follows from the type of the destination address, so it is not held by + `CEMIFlags` - see `CEMILData.address_type`. + """ + + INDIVIDUAL = 0 + GROUP = 1 + + def to_knx(self) -> int: + """Serialize to the Address Type bit of the control field - Ctrl2 b7.""" + return self << 7 + + @classmethod + def from_knx(cls, raw: int) -> CEMIAddressType: + """Parse the Address Type bit from the control field.""" + return cls(raw >> 7 & 0b1) + + +class CEMIFrameFormat(IntEnum): + """ + Extended Frame Format (EFF) of a cEMI L_Data frame. See 3/2/2 §2.2.5.3. + + `STANDARD` is used for L_Data_Standard frames as well as for long + L_Data_Extended frames. `LTE_HEE` covers 01xxb - the two least significant + bits hold the LTE extended address type. All other values are reserved. + """ + + STANDARD = 0b0000 + LTE_HEE = 0b0100 + + @classmethod + def _missing_(cls, value: object) -> CEMIFrameFormat | None: + """Resolve the whole 01xxb range to LTE_HEE; reserved values fail.""" + if isinstance(value, int) and value & 0b1100 == cls.LTE_HEE: + return cls.LTE_HEE + return None + + +@dataclass(slots=True) +class CEMIFlags: + """ + Control fields of a cEMI L_Data frame. + + `frame_type` and `frame_format` are informational: they hold what was received + and are ignored when serializing. The Frame Type of an outgoing frame follows + from its NPDU length and is passed to `to_knx()` by the frame; the Address Type + is not held at all - it can be read from the type of the destination address. + + `confirm_error` is only meaningful in an L_Data.con frame. + """ + + priority: CEMIPriority = CEMIPriority.LOW + # `repeat_on_error` and `system_broadcast` are inverted on the wire + repeat_on_error: bool = False + system_broadcast: bool = False + acknowledge_request: bool = False + confirm_error: bool = False + hop_count: int = 6 + # as received; not used when serializing + frame_type: CEMIFrameType = CEMIFrameType.STANDARD + frame_format: CEMIFrameFormat = CEMIFrameFormat.STANDARD + + def to_knx(self) -> int: + """ + Serialize the control field bits held by this object. + + The Frame Type and Address Type bits are left clear; the frame ORs them in + from `CEMIFrameType.to_knx()` and `CEMIAddressType.to_knx()`. `self.frame_type` + holds the Frame Type of a received frame and is not used here. + """ + if not 0 <= self.hop_count <= MAX_HOP_COUNT: + raise ConversionError(f"Hop count out of range: {self.hop_count}") + + return ( + (0 if self.repeat_on_error else DO_NOT_REPEAT) + | (0 if self.system_broadcast else BROADCAST) + | (self.priority << PRIORITY_OFFSET) + | (ACK_REQUESTED if self.acknowledge_request else 0) + | (CONFIRM_ERROR if self.confirm_error else 0) + | (self.hop_count << HOP_COUNT_OFFSET) + | CEMIFrameFormat.STANDARD + ) + + @classmethod + def from_knx(cls, raw: int) -> CEMIFlags: + """ + Parse the control field. + + Raise `ConversionError` for a reserved Extended Frame Format. The Frame Type + is kept but not evaluated: "any receiver shall be tolerant towards the use of + the Frame Format" - 3/6/3 §4.1.5.2.3. + """ + _eff = raw & EXTENDED_FRAME_FORMAT_MASK + try: + frame_format = CEMIFrameFormat(_eff) + except ValueError: + raise ConversionError( + f"Reserved Extended Frame Format: {_eff:#06b}" + ) from None + + return cls( + priority=CEMIPriority((raw & PRIORITY_MASK) >> PRIORITY_OFFSET), + repeat_on_error=not raw & DO_NOT_REPEAT, + system_broadcast=not raw & BROADCAST, + acknowledge_request=bool(raw & ACK_REQUESTED), + confirm_error=bool(raw & CONFIRM_ERROR), + hop_count=(raw & HOP_COUNT_MASK) >> HOP_COUNT_OFFSET, + frame_type=CEMIFrameType.from_knx(raw), + frame_format=frame_format, + ) + + def __str__(self) -> str: + """Return object as compact readable string.""" + _set_flags = [ + name + for name, value in ( + ("repeat_on_error", self.repeat_on_error), + ("system_broadcast", self.system_broadcast), + ("acknowledge_request", self.acknowledge_request), + ("confirm_error", self.confirm_error), + ) + if value + ] + return " ".join( + ( + self.priority.name, + self.frame_type.name, + f"hop_count={self.hop_count}", + *_set_flags, + ) + ) diff --git a/xknx/secure/data_secure.py b/xknx/secure/data_secure.py index 77d40e780..b82d24dab 100644 --- a/xknx/secure/data_secure.py +++ b/xknx/secure/data_secure.py @@ -212,7 +212,7 @@ def _received_secure_cemi( key=key, scf=s_apdu.scf, address_fields_raw=_address_fields_raw, - frame_flags=cemi_data.flags, + address_type=cemi_data.address_type, tpci=cemi_data.tpci, ) decrypted_payload = APCI.from_knx(plain_apdu_raw) @@ -260,7 +260,7 @@ def _secure_data_cemi( sequence_number=self.get_sequence_number(), address_fields_raw=cemi_data.src_addr.to_knx() + cemi_data.dst_addr.to_knx(), - frame_flags=cemi_data.flags, + address_type=cemi_data.address_type, tpci=cemi_data.tpci, ) secure_cemi_data = copy(cemi_data) diff --git a/xknx/secure/data_secure_asdu.py b/xknx/secure/data_secure_asdu.py index ae92a42e8..31fcfe5e0 100644 --- a/xknx/secure/data_secure_asdu.py +++ b/xknx/secure/data_secure_asdu.py @@ -4,6 +4,7 @@ from enum import IntEnum +from xknx.cemi.flags import CEMIAddressType from xknx.exceptions import DataSecureError from xknx.telegram.tpci import TPCI @@ -16,14 +17,16 @@ # Secure APCI is 0x03F1 - in block_0 it is used split into 2 octets _APCI_SEC_HIGH = 0x03 _APCI_SEC_LOW = 0xF1 -# only Address Type (IA / GA) and Extended Frame format are used -B0_AT_FIELD_FLAGS_MASK = 0b10001111 +# Of the cEMI control fields only the Address Type (IA / GA) and the Extended Frame +# Format are used; both live in Ctrl2. Ctrl1 - and with it the Frame Type - is not +# covered by the MAC. The Extended Frame Format is always 0 for supported frames, +# so `CEMIAddressType.to_knx()` is the whole octet. def block_0( sequence_number: bytes, address_fields_raw: bytes, - frame_flags: int, + address_type: CEMIAddressType, tpci_int: int, payload_length: int, ) -> bytes: @@ -34,7 +37,7 @@ def block_0( + bytes( ( 0, - frame_flags & B0_AT_FIELD_FLAGS_MASK, + address_type.to_knx(), (tpci_int << 2) + _APCI_SEC_HIGH, _APCI_SEC_LOW, 0, @@ -146,7 +149,7 @@ def init_from_plain_apdu( scf: SecurityControlField, sequence_number: int, address_fields_raw: bytes, - frame_flags: int, + address_type: CEMIAddressType, tpci: TPCI, ) -> SecureData: """Serialize to KNX raw data.""" @@ -159,7 +162,7 @@ def init_from_plain_apdu( block_0=block_0( sequence_number=sequence_number_bytes, address_fields_raw=address_fields_raw, - frame_flags=frame_flags, + address_type=address_type, tpci_int=tpci.to_knx(), payload_length=0, ), @@ -173,7 +176,7 @@ def init_from_plain_apdu( block_0=block_0( sequence_number=sequence_number_bytes, address_fields_raw=address_fields_raw, - frame_flags=frame_flags, + address_type=address_type, tpci_int=tpci.to_knx(), payload_length=len(apdu), ), @@ -218,7 +221,7 @@ def get_plain_apdu( key: bytes, scf: SecurityControlField, address_fields_raw: bytes, - frame_flags: int, + address_type: CEMIAddressType, tpci: TPCI, ) -> bytes: """ @@ -244,7 +247,7 @@ def get_plain_apdu( block_0=block_0( sequence_number=self.sequence_number_bytes, address_fields_raw=address_fields_raw, - frame_flags=frame_flags, + address_type=address_type, tpci_int=tpci.to_knx(), payload_length=len(dec_payload), ), @@ -260,7 +263,7 @@ def get_plain_apdu( block_0=block_0( sequence_number=self.sequence_number_bytes, address_fields_raw=address_fields_raw, - frame_flags=frame_flags, + address_type=address_type, tpci_int=tpci.to_knx(), payload_length=0, ),