forked from public/pysim
transport/smpp2sim: TERMINAL RESPONSE for proactive SEND SHORT MESSAGE
A multi part OTA response (full GP GET STATUS registry or some other fat response that exceeds one SMS) is delivered as several SMS via proactive SEND SHORT MESSAGE. The card only gives us another part if it receives a TERMINAL RESPONSE for the previous one. Currently smpp2sim Proact.handle_SendShortMessage relays the SMS but returns None, so the transport falls back to prepare_response(pcmd) with the ProactiveCommand collection (empty .children) and crashes with 'not enough values to unpack (expected 1, got 0)', dropps the SMPP link, and never fetches the remaining parts. Fix: make handle_SendShortMessage return a successful TERMINAL RESPONSE built from the decoded command so the handshake proceeds. prepare_response() is extended to handle collection via .decoded and raises a useful error. The send_apdu_checksw general_result='FIXME' path is now avoided for SendShortMessage but remains a problem for other handlers that return None. Change-Id: Ib96ce81c4ff093b8a6fc715f79617e95a2dd8433
This commit is contained in:
@@ -116,6 +116,14 @@ class Proact(ProactiveHandler):
|
||||
addr_ie.decoded['ton_npi']['numbering_plan_id'])
|
||||
logger.info(submit)
|
||||
self.send_sms_via_smpp(submit)
|
||||
# Return a successful TERMINAL RESPONSE.
|
||||
# This is important:
|
||||
# - without it the transport cannot complete the proactive command
|
||||
# - for a multi part OTA response, the card would never be asked to give us
|
||||
# the remaining SMS chunks.
|
||||
# 'pcmd' is a decoded SendShortMessage IE, which contains CommandDetails and
|
||||
# DeviceIdentities that prepare_response() echoes/inverts.
|
||||
return self.prepare_response(pcmd)
|
||||
|
||||
def handle_OpenChannel(self, pcmd: ProactiveCommand):
|
||||
"""Card requests opening a new channel via a UDP/TCP socket."""
|
||||
|
||||
@@ -70,10 +70,26 @@ class ProactiveHandler(abc.ABC):
|
||||
raise NotImplementedError('No handler method for %s' % pcmd.decoded)
|
||||
|
||||
def prepare_response(self, pcmd: ProactiveCommand, general_result: str = 'performed_successfully'):
|
||||
# TERMINAL RESPONSE per ETSI TS 102 223 section 6.8: Command details (6.8.1) echoed from the
|
||||
# command, Device identities (6.8.2) with source and destination swapped, Result (6.8.3).
|
||||
# pcmd can be
|
||||
# - decoded proactive command IE (.children contains CommandDetails/DeviceIdentities)
|
||||
# - ProactiveCommand collection wrapper (empty .children).
|
||||
# Normalise to the children obj, so both work:
|
||||
# - handler that passes its decoded command
|
||||
# - fallback path that passes collection
|
||||
children = list(getattr(pcmd, 'children', None) or [])
|
||||
if not any(isinstance(c, CommandDetails) for c in children):
|
||||
decoded = getattr(pcmd, 'decoded', None)
|
||||
if decoded is not None and decoded is not pcmd:
|
||||
children = list(getattr(decoded, 'children', None) or [])
|
||||
# The Command Details are echoed from the command that has been processed.
|
||||
(command_details,) = [c for c in pcmd.children if isinstance(c, CommandDetails)]
|
||||
command_details = next((c for c in children if isinstance(c, CommandDetails)), None)
|
||||
# invert the device identities
|
||||
(command_dev_ids,) = [c for c in pcmd.children if isinstance(c, DeviceIdentities)]
|
||||
command_dev_ids = next((c for c in children if isinstance(c, DeviceIdentities)), None)
|
||||
if command_details is None or command_dev_ids is None:
|
||||
raise ValueError('failed to prepare TERMINAL RESPONSE: proactive command has no '
|
||||
'CommandDetails/DeviceIdentities (%r)' % (pcmd,))
|
||||
rsp_dev_ids = DeviceIdentities()
|
||||
rsp_dev_ids.from_dict({'device_identities': {
|
||||
'dest_dev_id': command_dev_ids.decoded['source_dev_id'],
|
||||
|
||||
@@ -0,0 +1,77 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
"""Transport (as in t0/t1) tests"""
|
||||
|
||||
# (C) 2026 by sysmocom - s.f.m.c. GmbH
|
||||
# All Rights Reserved
|
||||
#
|
||||
# Author: Eric Wild <ewild@sysmocom.de>
|
||||
#
|
||||
# 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 unittest
|
||||
from osmocom.utils import h2b, b2h
|
||||
from pySim.cat import ProactiveCommand, CommandDetails, DeviceIdentities, Result
|
||||
from pySim.transport import ProactiveHandler
|
||||
|
||||
|
||||
def _send_short_message_pcmd():
|
||||
"""proactive SEND SHORT MESSAGE:
|
||||
D0 | CommandDetails(cmd 1, t 0x13, q 0) | DeviceIdentities(uicc->network)
|
||||
| dummy SMS_TPDU"""
|
||||
body = h2b('8103011300' + '82028183' + '8B04DEADBEEF')
|
||||
pdu = h2b('D0') + bytes([len(body)]) + body
|
||||
pcmd = ProactiveCommand()
|
||||
decoded = pcmd.from_tlv(pdu)
|
||||
return pcmd, decoded
|
||||
|
||||
|
||||
class Test_prepare_response(unittest.TestCase):
|
||||
"""TERMINAL RESPONSE.
|
||||
multi-part OTA response crash regression test."""
|
||||
|
||||
def setUp(self):
|
||||
self.h = ProactiveHandler.__new__(ProactiveHandler)
|
||||
|
||||
def test_on_decoded_command(self):
|
||||
_pcmd, decoded = _send_short_message_pcmd()
|
||||
til = self.h.prepare_response(decoded)
|
||||
self.assertEqual([type(c).__name__ for c in til],
|
||||
['CommandDetails', 'DeviceIdentities', 'Result'])
|
||||
# command details echoed, device id inverted, result OK
|
||||
self.assertEqual(b2h(til[0].to_tlv()), '8103011300')
|
||||
self.assertEqual(b2h(til[1].to_tlv()), '82028381')
|
||||
self.assertEqual(b2h(til[2].to_tlv()), '830100')
|
||||
|
||||
def test_on_collection_resolves_via_decoded(self):
|
||||
# Check that ProactiveCommand collection (empty .children) still works
|
||||
pcmd, _decoded = _send_short_message_pcmd()
|
||||
self.assertEqual(list(getattr(pcmd, 'children', []) or []), [])
|
||||
til = self.h.prepare_response(pcmd)
|
||||
self.assertEqual([type(c).__name__ for c in til],
|
||||
['CommandDetails', 'DeviceIdentities', 'Result'])
|
||||
self.assertEqual(b2h(til[0].to_tlv()), '8103011300')
|
||||
self.assertEqual(b2h(til[1].to_tlv()), '82028381')
|
||||
self.assertEqual(b2h(til[2].to_tlv()), '830100')
|
||||
|
||||
def test_missing_command_details_raises_clear_error(self):
|
||||
class _NoChildren:
|
||||
children = []
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
self.h.prepare_response(_NoChildren())
|
||||
self.assertIn('CommandDetails', str(ctx.exception))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user