diff --git a/contrib/smpp-ota-tool.py b/contrib/smpp-ota-tool.py index 36980014..885cf947 100755 --- a/contrib/smpp-ota-tool.py +++ b/contrib/smpp-ota-tool.py @@ -24,7 +24,8 @@ import smpplib.gsm import smpplib.client import smpplib.consts import time -from pySim.ota import OtaKeyset, OtaDialectSms, OtaAlgoCrypt, OtaAlgoAuth, CNTR_REQ, RC_CC_DS, POR_REQ +from pySim.ota import OtaKeyset, OtaDialectSms, OtaAlgoCrypt, OtaAlgoAuth, OtaCheckError, CNTR_REQ, RC_CC_DS, POR_REQ +from pySim.sms import ConcatenatedSmsReassembler from pySim.utils import b2h, h2b, is_hexstr from pathlib import Path @@ -108,28 +109,57 @@ class SmppHandler: self.tar = tar self.spi = spi self.remote_format = remote_format + self.reassembler = ConcatenatedSmsReassembler() def __del__(self): if self.client: self.client.unbind() self.client.disconnect() + def _decode_resp(self, tpud: bytes) -> tuple: + """Decode a response SMS-TPDU into (response_packet, decoded). + + Retry to decoding with ciphering disabled (in case the card has problems to decode the SMS-TDPU + we have sent, the response will contain an unencrypted error message) + """ + try: + return self.ota_dialect.decode_resp(self.ota_keyset, self.spi, tpud, + remote_format=self.remote_format) + except (ValueError, OtaCheckError): + spi = self.spi.copy() + spi['por_shall_be_ciphered'] = False + spi['por_rc_cc_ds'] = 'no_rc_cc_ds' + return self.ota_dialect.decode_resp(self.ota_keyset, spi, tpud, + remote_format=self.remote_format) + def message_received_handler(self, pdu): - if pdu.short_message: - logger.info("SMS-TPDU received: %s", b2h(pdu.short_message)) - try: - dec = self.ota_dialect.decode_resp(self.ota_keyset, self.spi, pdu.short_message, - remote_format=self.remote_format) - except ValueError: - # Retry to decoding with ciphering disabled (in case the card has problems to decode the SMS-TDPU - # we have sent, the response will contain an unencrypted error message) - spi = self.spi.copy() - spi['por_shall_be_ciphered'] = False - spi['por_rc_cc_ds'] = 'no_rc_cc_ds' - dec = self.ota_dialect.decode_resp(self.ota_keyset, spi, pdu.short_message, - remote_format=self.remote_format) - logger.info("SMS-TPDU decoded: %s", dec) - self.response = dec + if not pdu.short_message: + return None + logger.info("SMS-TPDU received: %s", b2h(pdu.short_message)) + tpud = self.reassembler.add(pdu.short_message) + if tpud is None: + logger.info("SMS-TPDU is part of concat message, waiting for more parts...") + return None + if tpud != pdu.short_message: + logger.info("SMS-TPDU reassembled: %s", b2h(tpud)) + try: + res, decoded = self._decode_resp(tpud) + except Exception as e: + # for example ENVELOPE POR + logger.warning("Ignoring undecodable resp SMS-TPDU (%s: %s)", type(e).__name__, e) + return None + logger.info("SMS-TPDU decoded: %s", (res, decoded)) + # large app response as reassembled SEND SHORT MESSAGE, but + # the ENVELOPE itself returns a POR without R-APDU. + # smpplib poll() drains all pending SMS in one call, so that PoR is processed + # right after the real response and would overwrite it, + # which leaves transceive_apdu with no last_response_data to return. + # Only allow a response that has no application data (decoded == None) + # if we do not already have a real one. + if decoded is None and self.response is not None and self.response[1] is not None: + logger.info("ignoring status response to keep earlier app response") + return None + self.response = (res, decoded) return None def message_sent_handler(self, pdu): diff --git a/pySim/sms.py b/pySim/sms.py index 62601b0e..75a9bde1 100644 --- a/pySim/sms.py +++ b/pySim/sms.py @@ -19,6 +19,7 @@ import typing import abc +import logging from bidict import bidict from construct import Int8ub, Byte, Bit, Flag, BitsInteger from construct import Struct, Enum, Tell, BitStruct, this, Padding @@ -28,6 +29,8 @@ from osmocom.utils import Hexstr, h2b, b2h from smpp.pdu import pdu_types, operations +logger = logging.getLogger(__name__) + BytesOrHex = typing.Union[Hexstr, bytes] class UserDataHeader: @@ -60,6 +63,109 @@ class UserDataHeader: return self._construct.build({'ies':self.ies, 'data':b''}) +class ConcatenatedSmsReassembler: + """3GPP TS 23.040 section 9.2.3.24 concat multi part reassembly + + A large user-data payload (e.g. a big OTA response packet) is split by the + sending entity into several SMS, + each carries a + - "concat short messages" IE in its UDH that identifies the set (ref num), + - total number of parts + - this parts seqno. + supports both: + IEI 0x00, section 9.2.3.24.1 8-bit ref form + IEI 0x08, section 9.2.3.24.8 the 16-bit ref form + + Feed each received TP-User-Data (UDH + payload) to add() which + returns the reassembled TP-User-Data once all parts of the set have arrived, + or None as long as parts are still missing. + + A non-concatenated SMS is returned unchanged, + just like one where the concat IE holds a reserved value: + TS 23.040 9.2.3.24.1 says + - both a total of zero + - a sequence number that is zero or greater than the total + that "the receiving entity shall ignore the whole IE", + we treat the message as a single, non-concatenated one and warn, not + as an error, so the caller does not die. + + The reassembled TP-User-Data is built with a UDH that contains + the non-concat IEs seen in the parts, for example the the OTA "response packet" + indicator IE 0x71, followed by the concatenated payloads in sequence order, + so exactly the single-SMS form the sender would have produced for a payload that fits + into one SMS. + This allows convenient decoding by the normal single part path.""" + + # IEI: Concatenated short messages, 8-bit reference number + # (see 3GPP TS 23.040 section 9.2.3.24 and section 9.2.3.24.1) + CONCAT_8BIT = 0x00 + # IEI: Concatenated short message, 16-bit reference number + # (see 3GPP TS 23.040 section 9.2.3.24 and section 9.2.3.24.8) + CONCAT_16BIT = 0x08 + + def __init__(self, max_sets: int = 8): + # keyed by (iei, ref, total): {'parts': {seq: payload}, 'header_ies'}, insertion ordered + self.sets = {} + self.max_sets = max_sets # incomplete sets kept, oldest is dropped beyond that + + @classmethod + def _parse_concat_ie(cls, ies) -> typing.Optional[typing.Tuple[int, int, int, int]]: + """Return (iei, ref, total, seq) of the concat IE, or None""" + for ie in ies: + if ie['iei'] == cls.CONCAT_8BIT and ie['length'] == 3: + v = ie['value'] + return cls.CONCAT_8BIT, v[0], v[1], v[2] + if ie['iei'] == cls.CONCAT_16BIT and ie['length'] == 4: + v = ie['value'] + return cls.CONCAT_16BIT, int.from_bytes(v[0:2], 'big'), v[2], v[3] + return None + + def add(self, tpud: BytesOrHex) -> typing.Optional[bytes]: + """Add one TP-User-Data. + Returns + - the reassembled TP-User-Data if set is complete or sms not multipart, + - else None""" + if isinstance(tpud, str): + tpud = h2b(tpud) + udh, payload = UserDataHeader.from_bytes(tpud) + concat = self._parse_concat_ie(udh.ies) + if concat is None: + return tpud + iei, ref, total, seq = concat + if total < 1 or seq < 1 or seq > total: + # TS 23.040 9.2.3.24.1 / 9.2.3.24.8, total zero or seqno zero / > total: + # Ignoring the IE means the message has no valid concat IE, which is a single part message. + # Better warn and hand it back rather than raise, so we don't kill the callers receive loop/session + logger.warning('Ignoring reserved concat IE (ref=%u total=%u seq=%u), treating the ' + 'message as non-concat', ref, total, seq) + return tpud + # TS 23.040 9.2.3.24.1 Total is constant in a set, refno only unique per IE form -> both set identity + # - full count = seqno 1..total is present + # - part disagreeing on the total ends up as set that cannot complete like set with missing parts + key = (iei, ref, total) + if key not in self.sets and len(self.sets) >= self.max_sets: + del self.sets[next(iter(self.sets))] + s = self.sets.setdefault(key, {'parts': {}, 'header_ies': []}) + s['parts'][seq] = payload + # - remember the non concat IEs (OTA 0x71 indicator for example) + # - keep first seen occurrence of each IEI, + # so app IE present only in the first segment is preserved independent of arrival order + seen = {ie['iei'] for ie in s['header_ies']} + for ie in udh.ies: + if ie['iei'] in (self.CONCAT_8BIT, self.CONCAT_16BIT): + continue + if ie['iei'] not in seen: + s['header_ies'].append(ie) + seen.add(ie['iei']) + if len(s['parts']) < total: + return None + # all parts present -> reassemble in seq order + del self.sets[(iei, ref, total)] + body = b''.join(s['parts'][i] for i in range(1, total + 1)) + header = UserDataHeader(s['header_ies']).to_bytes() + return header + body + + def smpp_dcs_is_8bit(dcs: pdu_types.DataCoding) -> bool: """Determine if the given SMPP data coding scheme is 8-bit or not.""" if dcs == pdu_types.DataCoding(pdu_types.DataCodingScheme.DEFAULT, diff --git a/tests/unittests/test_smpp_ota_tool.py b/tests/unittests/test_smpp_ota_tool.py new file mode 100644 index 00000000..ec60c854 --- /dev/null +++ b/tests/unittests/test_smpp_ota_tool.py @@ -0,0 +1,180 @@ +#!/usr/bin/env python3 +""" test for smpp-ota-tool SMS handling, specifically the multi part sms OTA response""" + +# (C) 2026 by sysmocom - s.f.m.c. GmbH +# All Rights Reserved +# +# Author: Eric Wild +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +import os.path +import importlib.util +import unittest + +from osmocom.utils import h2b, b2h + +from pySim.ota import OtaKeyset, OtaDialectSms, ExpandedRemoteResp +from pySim.sms import ConcatenatedSmsReassembler, UserDataHeader + +# import the hyphenated contrib script as a module to get at SmppHandler +# why do people name python files like that? why does everything have to be so hard? +_TOOL_PATH = os.path.join(os.path.dirname(__file__), '..', '..', 'contrib', 'smpp-ota-tool.py') +_spec = importlib.util.spec_from_file_location('smpp_ota_tool', _TOOL_PATH) +smpp_ota_tool = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(smpp_ota_tool) +SmppHandler = smpp_ota_tool.SmppHandler + + +class _FakePdu: + """Minimal mock for smpplib deliver_sm pdu.""" + def __init__(self, short_message): + self.short_message = short_message + + +class MultipartRelayTestCase(unittest.TestCase): + """message_received_handler must return the reassembled application + response and survive POR messages.""" + + # 3DES test keyset from tests/unittests/test_ota.py) used to make the + # handler happy. responses are plaintext, tests do not depend on keys. + def _handler(self, remote_format='expanded'): + h = object.__new__(SmppHandler) + h.client = None + h.ota_dialect = OtaDialectSms() + h.ota_keyset = OtaKeyset(algo_crypt='triple_des_cbc2', kic_idx=3, + kic=h2b('C21DD66ACAC13CB3BC8B331B24AFB57B'), + algo_auth='triple_des_cbc2', kid_idx=3, + kid=h2b('12110C78E678C25408233076AA033615')) + h.tar = h2b('000000') + # unciphered, no CC, PoR required + h.spi = {'counter': 'no_counter', 'ciphering': False, 'rc_cc_ds': 'no_rc_cc_ds', + 'por_in_submit': False, 'por': 'por_required', + 'por_shall_be_ciphered': False, 'por_rc_cc_ds': 'no_rc_cc_ds'} + h.remote_format = remote_format + h.reassembler = ConcatenatedSmsReassembler() + h.response = None + return h + + @staticmethod + def _plaintext_resp_sms(secured: bytes, sts: int = 0x00) -> bytes: + """Build a plaintext (unciphered, no-CC) OTA SMS response packet in the + canonical single-part form (UDH 02 71 00 + response packet).""" + rpl = 1 + 3 + 5 + 1 + 1 + len(secured) # RHL-STS + secured data + body = (rpl.to_bytes(2, 'big') + b'\x0a' + h2b('000000') + b'\x00' * 5 + + b'\x00' + bytes([sts]) + secured) + return b'\x02\x71\x00' + body + + @staticmethod + def _expanded_secured(response_data_hex: str, sw: str = '9000') -> bytes: + return ExpandedRemoteResp.build(dict(body=dict( + num_executed=dict(number_of_commands=1), + responses=[dict(r_apdu=dict(response_data=response_data_hex, status_word=sw))]))) + + @staticmethod + def _fragment_2(tpud: bytes, ref: int, first_len: int): + """Split 02 71 00 + body TP-UD into two SMS parts: + - part1 carries the OTA (0x71) IE + - part2 only concatenat IE + matches sja5 interaction""" + assert tpud[:3] == b'\x02\x71\x00' + body = tpud[3:] + ota_ie = {'iei': 0x71, 'length': 0, 'value': b''} + + def concat(seq): + return {'iei': 0x00, 'length': 3, 'value': bytes([ref, 2, seq])} + p1 = UserDataHeader([concat(1), ota_ie]).to_bytes() + body[:first_len] + p2 = UserDataHeader([concat(2)]).to_bytes() + body[first_len:] + return p1, p2 + + # ground truth: TP-User-Data captured from a sja5 + REAL_PART1 = h2b('070003010201710000e412000000df63afe4b06db21e2113be1be09e9b66f1c113ae841cca2d030064ec16b5b80ee5ce824604a4568109d25a82fb74a325df6f911bd0a4f858ece2c770039002c480269fc65953f5fd93ebbe528d97838bac4389a7303db2b073a37a9a1a51890457f41b49fc7905ce337e83449b65560501b8b845fe63339d557a928f2643') + REAL_PART2 = h2b('050003010202fd9c4e50ec40fb4427af518e9c08697405d91fbb6e9fa0b0935f48a560e15f2f3f27a2e44ef3a47280acce77f030fb70eb3df863c159177e2c0e3e53052fc7bb7ed171a491ded3ab7921861176a04305bc09fcf526c07bf6bb48a19e67cf18be5bc1') + REAL_REASSEMBLED = '02710000e412000000df63afe4b06db21e2113be1be09e9b66f1c113ae841cca2d030064ec16b5b80ee5ce824604a4568109d25a82fb74a325df6f911bd0a4f858ece2c770039002c480269fc65953f5fd93ebbe528d97838bac4389a7303db2b073a37a9a1a51890457f41b49fc7905ce337e83449b65560501b8b845fe63339d557a928f2643fd9c4e50ec40fb4427af518e9c08697405d91fbb6e9fa0b0935f48a560e15f2f3f27a2e44ef3a47280acce77f030fb70eb3df863c159177e2c0e3e53052fc7bb7ed171a491ded3ab7921861176a04305bc09fcf526c07bf6bb48a19e67cf18be5bc1' + + def test_real_card_parts_reassemble(self): + """two real card TP-UDs recombine into 233-byte single part packet: + UDH 02 71 00 + response packet""" + r = ConcatenatedSmsReassembler() + self.assertIsNone(r.add(self.REAL_PART1)) + out = r.add(self.REAL_PART2) + self.assertEqual(len(out), 233) + self.assertEqual(b2h(out), self.REAL_REASSEMBLED) + + def test_multipart_response_not_overwritten_by_por(self): + """reassembled application response must survive the ENVELOPE + trailing POR which contains no R-APDU""" + registry = bytes(range(198)) + app = self._plaintext_resp_sms(self._expanded_secured(b2h(registry))) + part1, part2 = self._fragment_2(app, ref=0x42, first_len=132) + # single part form must be too fat -> both parts must be concatenated + self.assertGreater(len(app), 140) + # ENVELOPE PoR: por_ok, but no app R-APDU + inline_por = self._plaintext_resp_sms(b'', sts=0x00) + + h = self._handler() + # arrival order + self.assertIsNone(h.message_received_handler(_FakePdu(part1))) + h.message_received_handler(_FakePdu(part2)) + h.message_received_handler(_FakePdu(inline_por)) + + # self.response must be app response, not the PoR! + self.assertIsNotNone(h.response) + res, decoded = h.response + self.assertEqual(res.response_status, 'por_ok') + self.assertIsNotNone(decoded) + self.assertEqual(decoded.last_response_data, b2h(registry)) + self.assertEqual(decoded.last_status_word, '9000') + + def test_undecodable_response_does_not_crash(self): + """response the handler can't decode must not escape out of the poll() + loop which would kill the tool, it must be ignored""" + # por_ok with a not expanded 'secured data' -> expanded parse raises + bad = self._plaintext_resp_sms(h2b('01612f'), sts=0x00) + h = self._handler(remote_format='expanded') + # must NOT raise + self.assertIsNone(h.message_received_handler(_FakePdu(bad))) + self.assertIsNone(h.response) + + def test_undecodable_por_after_good_response(self): + """real app response followed by undecodable PoR: + - good response is saved + - tool does not crash.""" + registry = bytes(range(120)) + app = self._plaintext_resp_sms(self._expanded_secured(b2h(registry))) + part1, part2 = self._fragment_2(app, ref=0x07, first_len=110) + bad_por = self._plaintext_resp_sms(h2b('deadbeef'), sts=0x00) + + h = self._handler() + h.message_received_handler(_FakePdu(part1)) + h.message_received_handler(_FakePdu(part2)) + self.assertIsNone(h.message_received_handler(_FakePdu(bad_por))) # no crash + res, decoded = h.response + self.assertIsNotNone(decoded) + self.assertEqual(decoded.last_response_data, b2h(registry)) + + def test_single_part_response_still_works(self): + """small response that fits one SMS turns into self.response, handled as before""" + h = self._handler() + sms = self._plaintext_resp_sms(self._expanded_secured('abcd', sw='9000')) + self.assertLessEqual(len(sms), 140) + h.message_received_handler(_FakePdu(sms)) + res, decoded = h.response + self.assertIsNotNone(decoded) + self.assertEqual(decoded.last_response_data, 'abcd') + self.assertEqual(decoded.last_status_word, '9000') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/unittests/test_sms.py b/tests/unittests/test_sms.py index d190528d..e82377a2 100644 --- a/tests/unittests/test_sms.py +++ b/tests/unittests/test_sms.py @@ -103,3 +103,126 @@ class Test_DELIVER(unittest.TestCase): self.assertEqual(d.tp_pid, 0x7f) self.assertEqual(d.tp_dcs, 0xf6) self.assertEqual(d.tp_udl, 8) + + +class Test_ConcatenatedSmsReassembler(unittest.TestCase): + """3GPP TS 23.040 9.2.3.24 reassembly of multi-part SMS. + + An OTA response that exceeds a single SHORT MESSAGE is delivered in several parts using + the SEND SHORT MESSAGE proactive command. The receiver must recombine the individual + parts into a single part before decoding.""" + + OTA_IE = {'iei': 0x71, 'length': 0, 'value': b''} + + @staticmethod + def _concat8(ref, tot, seq): + return {'iei': 0x00, 'length': 3, 'value': bytes([ref, tot, seq])} + + @staticmethod + def _concat16(ref, tot, seq): + return {'iei': 0x08, 'length': 4, 'value': ref.to_bytes(2, 'big') + bytes([tot, seq])} + + @staticmethod + def _part(ies, frag): + return UserDataHeader(ies).to_bytes() + frag + + def test_ground_truth_udh(self): + # part 1 UDH observed from sja5: 07 00 03 01 02 01 71 00 + built = self._part([self._concat8(1, 2, 1), self.OTA_IE], b'') + self.assertEqual(b2h(built), '0700030102017100') + + def test_ground_truth_udh_16bit(self): + # 9.2.3.24.8: 08 | 08 04 | 71 00 + built = self._part([self._concat16(0x1234, 2, 1), self.OTA_IE], b'') + self.assertEqual(b2h(built), '080804123402017100') + + def test_single_part_passthrough(self): + r = ConcatenatedSmsReassembler() + single = h2b('027100') + bytes(range(20)) + self.assertEqual(r.add(single), single) + + def test_two_part(self): + # second segment contains only the concat IE, no OTA IE + pkt = bytes(range(60)) + r = ConcatenatedSmsReassembler() + self.assertIsNone(r.add(self._part([self._concat8(1, 2, 1), self.OTA_IE], pkt[:35]))) + out = r.add(self._part([self._concat8(1, 2, 2)], pkt[35:])) + self.assertEqual(out, h2b('027100') + pkt) + + def test_out_of_order(self): + pkt = bytes(range(60)) + r = ConcatenatedSmsReassembler() + self.assertIsNone(r.add(self._part([self._concat8(5, 2, 2), self.OTA_IE], pkt[35:]))) + out = r.add(self._part([self._concat8(5, 2, 1), self.OTA_IE], pkt[:35])) + self.assertEqual(out, h2b('027100') + pkt) + + def test_three_part_out_of_order(self): + pkt = bytes(range(90)) + r = ConcatenatedSmsReassembler() + self.assertIsNone(r.add(self._part([self._concat8(7, 3, 3)], pkt[60:]))) + self.assertIsNone(r.add(self._part([self._concat8(7, 3, 1), self.OTA_IE], pkt[:30]))) + out = r.add(self._part([self._concat8(7, 3, 2)], pkt[30:60])) + self.assertEqual(out, h2b('027100') + pkt) + + def test_16bit_reference(self): + pkt = bytes(range(40)) + r = ConcatenatedSmsReassembler() + self.assertIsNone(r.add(self._part([self._concat16(0x1234, 2, 1), self.OTA_IE], pkt[:20]))) + out = r.add(self._part([self._concat16(0x1234, 2, 2)], pkt[20:])) + self.assertEqual(out, h2b('027100') + pkt) + + def test_interleaved_references(self): + # two concurrent concatenation sets at the same time + pkt = bytes(range(60)) + r = ConcatenatedSmsReassembler() + self.assertIsNone(r.add(self._part([self._concat8(1, 2, 1), self.OTA_IE], pkt[:35]))) + self.assertIsNone(r.add(self._part([self._concat8(9, 2, 1), self.OTA_IE], b'\xaa'))) + self.assertEqual(r.add(self._part([self._concat8(1, 2, 2)], pkt[35:])), h2b('027100') + pkt) + self.assertEqual(r.add(self._part([self._concat8(9, 2, 2)], b'\xbb')), h2b('027100') + b'\xaa\xbb') + + def test_reserved_concat_ie_is_ignored(self): + # TS 23.040 9.2.3.24.1: + # - a total of 0 + # - or a sequence number that is 0 or > total + # means "the receiving entity shall ignore the whole Information Element" + # the message is handed back unchanged as a single part msg and not rejected + # so the caller can handle the problem + r = ConcatenatedSmsReassembler() + for tot, seq in [(2, 3), # seq > total + (2, 0), # seq == 0 + (0, 1)]: # total == 0 + with self.subTest(total=tot, seq=seq): + part = self._part([self._concat8(1, tot, seq)], b'\x00') + self.assertEqual(r.add(part), part) + # nothing buffered so later valid set still reassembles properly + self.assertEqual(r.sets, {}) + pkt = bytes(range(40)) + self.assertIsNone(r.add(self._part([self._concat8(1, 2, 1), self.OTA_IE], pkt[:20]))) + self.assertEqual(r.add(self._part([self._concat8(1, 2, 2)], pkt[20:])), h2b('027100') + pkt) + + def test_inconsistent_totals_do_not_crash(self): + r = ConcatenatedSmsReassembler() + self.assertIsNone(r.add(self._part([self._concat8(1, 3, 3)], b'\x33'))) + self.assertIsNone(r.add(self._part([self._concat8(1, 2, 1)], b'\x11'))) + self.assertEqual(r.add(self._part([self._concat8(1, 2, 2)], b'\x22')), + h2b('00') + b'\x11\x22') # complete total=2 set + self.assertIn((0x00, 1, 3), r.sets) # total=3 set still waits + + def test_incomplete_sets_are_capped(self): + r = ConcatenatedSmsReassembler(max_sets=2) + for ref in (1, 2, 3): + self.assertIsNone(r.add(self._part([self._concat8(ref, 2, 1)], bytes([ref])))) + self.assertEqual(sorted(k[1] for k in r.sets), [2, 3]) # oldest evicted + self.assertIsNone(r.add(self._part([self._concat8(1, 2, 2)], b'\x11'))) + self.assertEqual(sorted(k[1] for k in r.sets), [1, 3]) + self.assertEqual(r.add(self._part([self._concat8(3, 2, 2)], b'\x33')), h2b('00') + b'\x03\x33') + + def test_same_reference_in_both_ie_forms(self): + # the refno only unique per IE form (9.2.3.24.1 vs .8) -> two sets + r = ConcatenatedSmsReassembler() + self.assertIsNone(r.add(self._part([self._concat8(1, 2, 1)], b'\x0a'))) + self.assertIsNone(r.add(self._part([self._concat16(1, 2, 2)], b'\x1b'))) + self.assertEqual(r.add(self._part([self._concat16(1, 2, 1)], b'\x0b')), + h2b('00') + b'\x0b\x1b') + self.assertEqual(r.add(self._part([self._concat8(1, 2, 2)], b'\x1a')), + h2b('00') + b'\x0a\x1a')