mirror of
https://github.com/smittix/intercept.git
synced 2026-07-05 16:18:12 -07:00
fix: harden DSC decoder against noise-induced false decodes
Stricter dot pattern detection (200 bits/100 alternations), bounded phasing strip (max 7), symbol check bit parity validation, EOS minimum position check, strict MMSI decode (reject out-of-range symbols), format-aware telecommand extraction, and expanded critical category detection. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Smittix
+114
-13
@@ -515,18 +515,16 @@ class TestDSCDecoder:
|
||||
assert result == '002320001'
|
||||
|
||||
def test_decode_mmsi_short_symbols(self, decoder):
|
||||
"""Test MMSI decoding handles short symbol list."""
|
||||
"""Test MMSI decoding returns None for short symbol list."""
|
||||
result = decoder._decode_mmsi([1, 2, 3])
|
||||
assert result == '000000000'
|
||||
assert result is None
|
||||
|
||||
def test_decode_mmsi_invalid_symbols(self, decoder):
|
||||
"""Test MMSI decoding handles invalid symbol values."""
|
||||
# Symbols > 99 should be treated as 0
|
||||
"""Test MMSI decoding returns None for out-of-range symbols."""
|
||||
# Symbols > 99 should cause decode to fail
|
||||
symbols = [100, 32, 12, 34, 56]
|
||||
result = decoder._decode_mmsi(symbols)
|
||||
# First symbol (100) becomes 00, padded result "0032123456",
|
||||
# trim leading pad digit -> "032123456"
|
||||
assert result == '032123456'
|
||||
assert result is None
|
||||
|
||||
def test_decode_position_northeast(self, decoder):
|
||||
"""Test position decoding for NE quadrant."""
|
||||
@@ -577,8 +575,9 @@ class TestDSCDecoder:
|
||||
def test_bits_to_symbol(self, decoder):
|
||||
"""Test bit to symbol conversion."""
|
||||
# Symbol value is first 7 bits (LSB first)
|
||||
# Value 100 = 0b1100100 -> bits [0,0,1,0,0,1,1, x,x,x]
|
||||
bits = [0, 0, 1, 0, 0, 1, 1, 0, 0, 0]
|
||||
# Value 100 = 0b1100100 -> bits [0,0,1,0,0,1,1] -> 3 ones
|
||||
# Check bits must make total even -> need 1 more one -> [1,0,0]
|
||||
bits = [0, 0, 1, 0, 0, 1, 1, 1, 0, 0]
|
||||
result = decoder._bits_to_symbol(bits)
|
||||
assert result == 100
|
||||
|
||||
@@ -588,14 +587,14 @@ class TestDSCDecoder:
|
||||
assert result == -1
|
||||
|
||||
def test_detect_dot_pattern(self, decoder):
|
||||
"""Test dot pattern detection."""
|
||||
# Dot pattern is alternating 1010101...
|
||||
decoder.bit_buffer = [1, 0] * 25 # 50 alternating bits
|
||||
"""Test dot pattern detection with 200+ alternating bits."""
|
||||
# Dot pattern requires at least 200 bits / 100 alternations
|
||||
decoder.bit_buffer = [1, 0] * 110 # 220 alternating bits
|
||||
assert decoder._detect_dot_pattern() is True
|
||||
|
||||
def test_detect_dot_pattern_insufficient(self, decoder):
|
||||
"""Test dot pattern not detected with insufficient alternations."""
|
||||
decoder.bit_buffer = [1, 0] * 5 # Only 10 bits
|
||||
decoder.bit_buffer = [1, 0] * 40 # Only 80 bits, below 200 threshold
|
||||
assert decoder._detect_dot_pattern() is False
|
||||
|
||||
def test_detect_dot_pattern_not_alternating(self, decoder):
|
||||
@@ -603,6 +602,84 @@ class TestDSCDecoder:
|
||||
decoder.bit_buffer = [1, 1, 1, 1, 0, 0, 0, 0] * 5
|
||||
assert decoder._detect_dot_pattern() is False
|
||||
|
||||
def test_bounded_phasing_strip(self, decoder):
|
||||
"""Test that >7 phasing symbols causes decode to return None."""
|
||||
# Build message bits: 10 phasing symbols (120) + format + data
|
||||
# Each symbol is 10 bits. Phasing symbol 120 = 0b1111000 LSB first
|
||||
# 120 in 7 bits LSB-first: 0,0,0,1,1,1,1 + 3 check bits
|
||||
# 120 = 0b1111000 -> LSB first: 0,0,0,1,1,1,1 -> ones=4 (even) -> check [0,0,0]
|
||||
phasing_bits = [0, 0, 0, 1, 1, 1, 1, 0, 0, 0] # symbol 120
|
||||
# 10 phasing symbols (>7 max)
|
||||
decoder.message_bits = phasing_bits * 10
|
||||
# Add some non-phasing symbols after (enough for a message)
|
||||
# Symbol 112 (INDIVIDUAL) = 0b1110000 LSB-first: 0,0,0,0,1,1,1 -> ones=3 (odd) -> need odd check
|
||||
# For simplicity, just add enough bits for the decoder to attempt
|
||||
for _ in range(20):
|
||||
decoder.message_bits.extend([0, 0, 0, 0, 1, 1, 1, 1, 0, 0])
|
||||
result = decoder._try_decode_message()
|
||||
assert result is None
|
||||
|
||||
def test_eos_minimum_length(self, decoder):
|
||||
"""Test that EOS found too early in the symbol stream is skipped."""
|
||||
# Build a message where EOS appears at position 5 (< MIN_SYMBOLS_FOR_FORMAT=12)
|
||||
# This should not be accepted as a valid message end
|
||||
# Symbol 127 (EOS) = 0b1111111 LSB-first: 1,1,1,1,1,1,1 -> ones=7 (odd) -> check needs 1 one
|
||||
# Use a simple approach: create symbols directly via _try_decode_message
|
||||
# Create 5 normal symbols + EOS at position 5 — should be skipped
|
||||
# Followed by more symbols and a real EOS at position 15
|
||||
from utils.dsc.decoder import DSCDecoder
|
||||
d = DSCDecoder()
|
||||
|
||||
# Build symbols manually: we need _try_decode_message to find EOS too early
|
||||
# Symbol 112 = format code. We'll build 10 bits per symbol.
|
||||
# Since check bit validation is now active, we need valid check bits.
|
||||
# Symbol value 10 = 0b0001010 LSB-first: 0,1,0,1,0,0,0, ones=2 (even) -> check [0,0,0]
|
||||
sym_10 = [0, 1, 0, 1, 0, 0, 0, 0, 0, 0]
|
||||
# Symbol 127 (EOS) = 0b1111111, ones=7 (odd) -> check needs odd total -> [1,0,0]
|
||||
sym_eos = [1, 1, 1, 1, 1, 1, 1, 1, 0, 0]
|
||||
|
||||
# 5 normal symbols + early EOS (should be skipped) + 8 more normal + real EOS
|
||||
d.message_bits = sym_10 * 5 + sym_eos + sym_10 * 8 + sym_eos
|
||||
result = d._try_decode_message()
|
||||
# The early EOS at index 5 should be skipped; the one at index 14
|
||||
# is past MIN_SYMBOLS_FOR_FORMAT so it can be accepted.
|
||||
# But the message content is garbage, so _decode_symbols will likely
|
||||
# return None for other reasons. The key test: it doesn't return a
|
||||
# message truncated at position 5.
|
||||
# Just verify no crash and either None or a valid longer message
|
||||
# (not truncated at the early EOS)
|
||||
assert result is None or len(result.get('raw', '')) > 18
|
||||
|
||||
def test_bits_to_symbol_check_bit_validation(self, decoder):
|
||||
"""Test that _bits_to_symbol rejects symbols with invalid check bits."""
|
||||
# Symbol 100 = 0b1100100 LSB-first: 0,0,1,0,0,1,1
|
||||
# ones in data = 3, need total even -> check bits need 1 one
|
||||
# Valid: [0,0,1,0,0,1,1, 1,0,0] -> total ones = 4 (even) -> valid
|
||||
valid_bits = [0, 0, 1, 0, 0, 1, 1, 1, 0, 0]
|
||||
assert decoder._bits_to_symbol(valid_bits) == 100
|
||||
|
||||
# Invalid: flip one check bit -> total ones = 5 (odd) -> invalid
|
||||
invalid_bits = [0, 0, 1, 0, 0, 1, 1, 0, 0, 0]
|
||||
assert decoder._bits_to_symbol(invalid_bits) == -1
|
||||
|
||||
def test_safety_is_critical(self):
|
||||
"""Test that SAFETY category is marked as critical."""
|
||||
import json
|
||||
|
||||
from utils.dsc.parser import parse_dsc_message
|
||||
|
||||
raw = json.dumps({
|
||||
'type': 'dsc',
|
||||
'format': 123,
|
||||
'source_mmsi': '232123456',
|
||||
'category': 'SAFETY',
|
||||
'timestamp': '2025-01-15T12:00:00Z',
|
||||
'raw': '123232123456100122',
|
||||
})
|
||||
msg = parse_dsc_message(raw)
|
||||
assert msg is not None
|
||||
assert msg['is_critical'] is True
|
||||
|
||||
|
||||
class TestDSCConstants:
|
||||
"""Tests for DSC constants."""
|
||||
@@ -670,3 +747,27 @@ class TestDSCConstants:
|
||||
assert DSC_BAUD_RATE == 1200
|
||||
assert DSC_MARK_FREQ == 2100
|
||||
assert DSC_SPACE_FREQ == 1300
|
||||
|
||||
def test_telecommand_codes_full(self):
|
||||
"""Test TELECOMMAND_CODES_FULL covers 0-127 range."""
|
||||
from utils.dsc.constants import TELECOMMAND_CODES_FULL
|
||||
|
||||
assert len(TELECOMMAND_CODES_FULL) == 128
|
||||
# Known codes map correctly
|
||||
assert TELECOMMAND_CODES_FULL[100] == 'F3E_G3E_ALL'
|
||||
assert TELECOMMAND_CODES_FULL[107] == 'DISTRESS_ACK'
|
||||
# Unknown codes map to "UNKNOWN"
|
||||
assert TELECOMMAND_CODES_FULL[0] == 'UNKNOWN'
|
||||
assert TELECOMMAND_CODES_FULL[99] == 'UNKNOWN'
|
||||
|
||||
def test_telecommand_formats(self):
|
||||
"""Test TELECOMMAND_FORMATS contains correct format codes."""
|
||||
from utils.dsc.constants import TELECOMMAND_FORMATS
|
||||
|
||||
assert {112, 114, 116, 120, 123} == TELECOMMAND_FORMATS
|
||||
|
||||
def test_min_symbols_for_format(self):
|
||||
"""Test MIN_SYMBOLS_FOR_FORMAT constant."""
|
||||
from utils.dsc.constants import MIN_SYMBOLS_FOR_FORMAT
|
||||
|
||||
assert MIN_SYMBOLS_FOR_FORMAT == 12
|
||||
|
||||
Reference in New Issue
Block a user