diff --git a/bellows/ash.py b/bellows/ash.py index e278006a..9b4532b2 100644 --- a/bellows/ash.py +++ b/bellows/ash.py @@ -366,6 +366,7 @@ def __init__(self, ezsp_protocol) -> None: self._ncp_reset_code: t.NcpResetCode | None = None self._ncp_state: NcpState = NcpState.CONNECTED + self._in_reject_condition: bool = False def connection_made(self, transport): self._transport = transport @@ -393,6 +394,22 @@ def close(self): self._transport.close() self._transport = None + def _reject_frame(self) -> None: + """Send NAK and enter reject condition if not already rejected.""" + if self._ncp_state == NcpState.CONNECTED and not self._in_reject_condition: + _LOGGER.debug("Entering reject condition, sending NAK") + with contextlib.suppress(NcpFailure): + self._write_frame(NakFrame(res=0, ncp_ready=0, ack_num=self._rx_seq)) + self._in_reject_condition = True + elif self._in_reject_condition: + _LOGGER.debug("Already in reject condition, suppressing NAK") + + def _clear_reject_condition(self) -> None: + """Clear reject condition and allow future error responses.""" + if self._in_reject_condition: + _LOGGER.debug("Clearing reject condition") + self._in_reject_condition = False + @staticmethod def _stuff_bytes(data: bytes) -> bytes: """Stuff bytes for transmission""" @@ -473,11 +490,12 @@ def data_received(self, data: bytes) -> None: "Failed to parse frame %r", frame_bytes, exc_info=True ) - with contextlib.suppress(NcpFailure): - self._write_frame( - NakFrame(res=0, ncp_ready=0, ack_num=self._rx_seq), - prefix=(Reserved.CANCEL,), - ) + # Reject DATA frame parse errors (SDK behavior) + if ( + len(data) > 0 + and (data[0] & DataFrame.MASK) == DataFrame.MASK_VALUE + ): + self._reject_frame() else: self.frame_received(frame) elif reserved_byte == Reserved.CANCEL: @@ -544,14 +562,18 @@ def data_frame_received(self, frame: DataFrame) -> None: self._rx_seq = (frame.frm_num + 1) % 8 self._write_frame(AckFrame(res=0, ncp_ready=0, ack_num=self._rx_seq)) + # Clear reject condition on valid in-sequence DATA frame + self._clear_reject_condition() + self._ezsp_protocol.data_received(frame.ezsp_frame) elif frame.re_tx: # Retransmitted frames must be immediately ACKed even if they are out of # sequence self._write_frame(AckFrame(res=0, ncp_ready=0, ack_num=self._rx_seq)) else: - _LOGGER.debug("Received an out of sequence frame: %r", frame) - self._write_frame(NakFrame(res=0, ncp_ready=0, ack_num=self._rx_seq)) + # Out-of-sequence non-retransmitted frame: reject it + _LOGGER.debug("Received out-of-sequence frame: %r", frame) + self._reject_frame() def rstack_frame_received(self, frame: RStackFrame) -> None: self._ncp_reset_code = None @@ -559,6 +581,7 @@ def rstack_frame_received(self, frame: RStackFrame) -> None: self._tx_seq = 0 self._rx_seq = 0 + self._clear_reject_condition() self._cancel_pending_data_frames(NcpFailure(code=frame.reset_code)) self._change_ack_timeout(T_RX_ACK_INIT) self._ezsp_protocol.reset_received(frame.reset_code) @@ -572,6 +595,7 @@ def nak_frame_received(self, frame: NakFrame) -> None: def rst_frame_received(self, frame: RstFrame) -> None: self._ncp_reset_code = None self._ncp_state = NcpState.CONNECTED + self._clear_reject_condition() def error_frame_received(self, frame: ErrorFrame) -> None: _LOGGER.debug("NCP has entered failed state: %s", frame.reset_code) diff --git a/tests/test_ash.py b/tests/test_ash.py index e1071815..f97b2c4f 100644 --- a/tests/test_ash.py +++ b/tests/test_ash.py @@ -639,6 +639,193 @@ async def test_rstack_cancels_pending_frames() -> None: assert exc_info.value.code == t.NcpResetCode.RESET_POWER_ON +async def test_reject_condition_prevents_nak_amplification(caplog) -> None: + """Test that reject condition prevents NAK amplification on repeated errors.""" + caplog.set_level(logging.DEBUG) + + ezsp = MagicMock() + protocol = ash.AshProtocol(ezsp) + transport = MagicMock() + transport.is_closing.return_value = False + protocol.connection_made(transport) + + # Start connected + protocol._ncp_state = ash.NcpState.CONNECTED + + # Send first bad DATA frame with invalid CRC (control byte 0x00 = DATA frame) + bad_frame_1 = b"\x00Some bad data\xDE\xAD" # Invalid CRC + protocol.data_received(bad_frame_1 + bytes([ash.Reserved.FLAG])) + + # Should enter reject condition and send one NAK + assert "Entering reject condition" in caplog.text + nak_calls = [ + c + for c in transport.write.mock_calls + if c.args and (c.args[0][0] & ash.NakFrame.MASK) == ash.NakFrame.MASK_VALUE + ] + assert len(nak_calls) == 1 + + caplog.clear() + transport.write.reset_mock() + + # Send second bad DATA frame with invalid CRC + bad_frame_2 = b"\x00More bad data\xBE\xEF" # Invalid CRC + protocol.data_received(bad_frame_2 + bytes([ash.Reserved.FLAG])) + + # Should suppress NAK due to reject condition + assert "Already in reject condition" in caplog.text + assert len(transport.write.mock_calls) == 0 + + caplog.clear() + + # Send third bad DATA frame with invalid CRC + bad_frame_3 = b"\x00Even more bad data\xCA\xFE" # Invalid CRC + protocol.data_received(bad_frame_3 + bytes([ash.Reserved.FLAG])) + + # Still suppressing + assert "Already in reject condition" in caplog.text + assert len(transport.write.mock_calls) == 0 + + caplog.clear() + + # Send valid in-sequence DATA frame to clear reject condition + protocol._rx_seq = 0 + good_frame = ash.DataFrame(frm_num=0, re_tx=0, ack_num=0, ezsp_frame=b"good") + protocol.data_received(good_frame.to_bytes() + bytes([ash.Reserved.FLAG])) + + # Should clear reject condition and send ACK + assert "Clearing reject condition" in caplog.text + assert protocol._in_reject_condition is False + + caplog.clear() + transport.write.reset_mock() + + # Now another bad frame should trigger reject condition again + bad_frame_4 = b"\x00Bad again\xFA\xDE" # Invalid CRC + protocol.data_received(bad_frame_4 + bytes([ash.Reserved.FLAG])) + + # Should enter reject condition again and send NAK + assert "Entering reject condition" in caplog.text + nak_calls = [ + c + for c in transport.write.mock_calls + if c.args and (c.args[0][0] & ash.NakFrame.MASK) == ash.NakFrame.MASK_VALUE + ] + assert len(nak_calls) == 1 + + +async def test_reject_condition_out_of_sequence_frames(caplog) -> None: + """Test that reject condition prevents NAK amplification on out-of-sequence frames.""" + caplog.set_level(logging.DEBUG) + + ezsp = MagicMock() + protocol = ash.AshProtocol(ezsp) + transport = MagicMock() + transport.is_closing.return_value = False + protocol.connection_made(transport) + + protocol._ncp_state = ash.NcpState.CONNECTED + protocol._rx_seq = 3 # Expecting frame 3 + + # Send out-of-sequence frame 4 (not retransmitted) + frame_4 = ash.DataFrame(frm_num=4, re_tx=0, ack_num=0, ezsp_frame=b"frame 4") + protocol.data_received(frame_4.to_bytes() + bytes([ash.Reserved.FLAG])) + + # Should enter reject condition and send NAK + assert "Entering reject condition" in caplog.text + assert protocol._in_reject_condition is True + nak_calls = [ + c + for c in transport.write.mock_calls + if c.args and (c.args[0][0] & ash.NakFrame.MASK) == ash.NakFrame.MASK_VALUE + ] + assert len(nak_calls) == 1 + + caplog.clear() + transport.write.reset_mock() + + # Send another out-of-sequence frame 5 + frame_5 = ash.DataFrame(frm_num=5, re_tx=0, ack_num=0, ezsp_frame=b"frame 5") + protocol.data_received(frame_5.to_bytes() + bytes([ash.Reserved.FLAG])) + + # Should suppress NAK due to reject condition + assert "Already in reject condition" in caplog.text + assert len(transport.write.mock_calls) == 0 + + # Send another out-of-sequence frame 6 + frame_6 = ash.DataFrame(frm_num=6, re_tx=0, ack_num=0, ezsp_frame=b"frame 6") + protocol.data_received(frame_6.to_bytes() + bytes([ash.Reserved.FLAG])) + + # Still suppressing + assert len(transport.write.mock_calls) == 0 + + caplog.clear() + + # Finally send the expected frame 3 + frame_3 = ash.DataFrame(frm_num=3, re_tx=0, ack_num=0, ezsp_frame=b"frame 3") + protocol.data_received(frame_3.to_bytes() + bytes([ash.Reserved.FLAG])) + + # Should clear reject condition + assert "Clearing reject condition" in caplog.text + assert protocol._in_reject_condition is False + + +async def test_reject_condition_retransmitted_frames_always_acked() -> None: + """Test that retransmitted out-of-sequence frames are ACKed without entering reject.""" + ezsp = MagicMock() + protocol = ash.AshProtocol(ezsp) + transport = MagicMock() + transport.is_closing.return_value = False + protocol.connection_made(transport) + + protocol._ncp_state = ash.NcpState.CONNECTED + protocol._rx_seq = 3 # Expecting frame 3 + + # Send out-of-sequence retransmitted frame (re_tx=1) + retx_frame = ash.DataFrame(frm_num=4, re_tx=1, ack_num=0, ezsp_frame=b"retx") + protocol.data_received(retx_frame.to_bytes() + bytes([ash.Reserved.FLAG])) + + # Should send ACK, not NAK, and not enter reject condition + assert protocol._in_reject_condition is False + ack_calls = [ + c + for c in transport.write.mock_calls + if c.args and (c.args[0][0] & ash.AckFrame.MASK) == ash.AckFrame.MASK_VALUE + ] + assert len(ack_calls) >= 1 + + +async def test_reject_condition_firmware_amplification_bug() -> None: + """Test reject condition prevents OpenThread RCP amplification.""" + ezsp = MagicMock() + protocol = ash.AshProtocol(ezsp) + transport = MagicMock() + transport.is_closing.return_value = False + protocol.connection_made(transport) + + protocol._ncp_state = ash.NcpState.CONNECTED + + # Simulate firmware sending garbage that happens to end with ~ + protocol.data_received(b"\x00\x06p1A A0 54 1A]\n\x07K~") + + # ASH enters reject condition and sends a NAK + assert len(transport.write.mock_calls) == 1 + + transport.write.reset_mock() + + # Firmware responds to the NAK with yet another corrupt frame + protocol.data_received(b"\x00\x06pFraming error 6: [\xca\xfd~") + + # We no longer reply, as we are in a reject condition + assert len(transport.write.mock_calls) == 0 + + # Firmware sends yet another corrupt response + protocol.data_received(b"\x00\x06p1A A0 54 1A]\n\x07K~") + + # Still suppressing - no amplification + assert len(transport.write.mock_calls) == 0 + + def test_ncp_failure_comparison() -> None: exc1 = ash.NcpFailure(code=t.NcpResetCode.ERROR_EXCEEDED_MAXIMUM_ACK_TIMEOUT_COUNT) exc2 = ash.NcpFailure(code=t.NcpResetCode.RESET_POWER_ON)