Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/43544?usp=email )
Change subject: sms/smpp-ota-tool: reassemble multi part response SMS ......................................................................
sms/smpp-ota-tool: reassemble multi part response SMS
A large OTA response (for example GP GET STATUS app registry) is split by the card into multiple SMS, each carries a TS 23.040 9.2.3.24 'concatenated short messages' IE in its UDH. Nothing recombines them, so smpp-ota-tool currently only sees the first incomplete part.
Add ConcatenatedSmsReassembler that accepts TP-User-Data, buffers parts by reference number, and returns the reassembled TP-User-Data in the canonical single-part form. Non-concatenated SMS pass through unchanged. Both the 8-bit+16-bit references are supported.
A reserved value in the concatenation IE is not an error, these messages are handed back as is rather than rejected, so the caller can deal with that.
Reassembly leads to a result that looks like a fat single part message the card could have produced given infinite sms sizes, so the existing decode_resp path is unaffected.
Feeding those parts to the ota tool needs two more fixes because the card returns the application response as several SMS via proactive SEND SHORT MESSAGE while the ENVELOPE SMS-PP DOWNLOAD itself contains the POR without a app R-APDU. smpplib poll() drains everything, so message_received_handler runs on each:
- a later status-only response must not overwrite an application response already captured, or transceive_apdu finds no last_response_data and raises - a response that cannot be decoded is be logged and skipped rather raised out of client.poll() which kills the tool.
Plus tests from a sja5 session.
Change-Id: I8c81097e607e0d055c4f031bbcc8a74d5c24a0e7 --- M contrib/smpp-ota-tool.py M pySim/sms.py A tests/unittests/test_smpp_ota_tool.py M tests/unittests/test_sms.py 4 files changed, 433 insertions(+), 16 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/44/43544/1
diff --git a/contrib/smpp-ota-tool.py b/contrib/smpp-ota-tool.py index 3698001..885cf94 100755 --- a/contrib/smpp-ota-tool.py +++ b/contrib/smpp-ota-tool.py @@ -24,7 +24,8 @@ 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 @@ 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 62601b0..0c73a53 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 smpp.pdu import pdu_types, operations
+logger = logging.getLogger(__name__) + BytesOrHex = typing.Union[Hexstr, bytes]
class UserDataHeader: @@ -60,6 +63,102 @@ 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.""" + + # TS 23.040 9.2.3.24.1/.8 IEI of the concat IE + CONCAT_8BIT = 0x00 + CONCAT_16BIT = 0x08 + + def __init__(self): + # keyed by (iei, ref, total): {'parts': {seq: payload}, 'header_ies'} + self.sets = {} + + @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.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 + s = self.sets.setdefault((iei, ref, total), {'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 0000000..80fb008 --- /dev/null +++ b/tests/unittests/test_smpp_ota_tool.py @@ -0,0 +1,179 @@ +#!/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 info@sysmocom.de +# +# 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 http://www.gnu.org/licenses/. + +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 d190528..5984f78 100644 --- a/tests/unittests/test_sms.py +++ b/tests/unittests/test_sms.py @@ -103,3 +103,112 @@ self.assertEqual(d.tp_pid, 0x7f) self.assertEqual(d.tp_dcs, 0xf6) self.assertEqual(d.tp_udl, 8) + + +class Test_ConcatenatedSmsReassembler(unittest.TestCase): + """TS 23.040 9.2.3.24 reassembly of multi-part SMS. + + OTA response that excees a single SMS is delivered as several parts via + proactive SEND SHORT MESSAGE, reassembler must recombine into single part + form 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_single_part_passthrough(self): + r = ConcatenatedSmsReassembler() + single = h2b('027100') + bytes(range(20)) + self.assertEqual(r.add(single), single) + + 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_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_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')