08 Sep '26
Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/43566?usp=email )
Change subject: contrib: add scp81 OTA trigger builder
......................................................................
contrib: add scp81 OTA trigger builder
Change-Id: Ic646626319cd7a26195d0ffff55900032abcddeb
---
A contrib/scp81_trigger.py
1 file changed, 99 insertions(+), 0 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/66/43566/1
diff --git a/contrib/scp81_trigger.py b/contrib/scp81_trigger.py
new file mode 100644
index 0000000..00cd269
--- /dev/null
+++ b/contrib/scp81_trigger.py
@@ -0,0 +1,99 @@
+#!/usr/bin/env python3
+"""scp81_trigger.py -- build the OTA packet that asks the card to open an SCP81 admin session."""
+
+# (C) 2026 by sysmocom - s.f.m.c. GmbH <info(a)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/>.
+
+
+# Prints the apdu line for AdmSessTriggerParams TLV as the sms secured data, Expanded RFM mode,
+# to be fed into pysim_shell.py
+#
+# security params supplied either
+# - in the trigger
+# - from the cards data object,
+# trigger wins when both are present.
+# --no-sec omits them from the trigger so the stored ones are used.
+#
+# example params:
+# --psk-id 'PSK Identity 123' --kvn 0x41 --kid-ref 5
+# --ip 127.0.0.1 --port 8080 --buffer 512
+# --host 172.96.0.1 --uri '/server/adminagent?cmd=1'
+
+import argparse
+import sys
+
+from osmocom.utils import b2h # noqa: E402
+from pySim.cat import (sms_pp_download_envelope, BearerDescription, # noqa: E402
+ BufferSize, UiccTransportLevel, OtherAddress)
+from pySim.global_platform.http import (AdmSessTriggerParams, AdmSessionParams, # noqa: E402
+ SecurityParams, HttpPostParams, RasConnectionParams,
+ AdminHostParam, AdminUriParam)
+
+
+def main():
+ ap = argparse.ArgumentParser(description=__doc__,
+ formatter_class=argparse.RawDescriptionHelpFormatter)
+ ap.add_argument("--psk-id", help="PSK identity for ClientHello (required unless --no-sec)")
+ ap.add_argument("--kvn", type=lambda s: int(s, 0), help="key version of the PSK (required unless --no-sec)")
+ ap.add_argument("--kid-ref", type=lambda s: int(s, 0), help="PSK key id (required unless --no-sec)")
+ ap.add_argument("--host", help="HTTP Host header (required unless --no-http)")
+ ap.add_argument("--uri", help="HTTP request URI (required unless --no-http)")
+ ap.add_argument("--ip", help="administration server address, BIP (required unless --no-conn)")
+ ap.add_argument("--port", type=int, help="administration server port (required unless --no-conn)")
+ ap.add_argument("--buffer", type=int, help="BIP buffer size (required unless --no-conn)")
+ ap.add_argument("--no-conn", action="store_true", help="omit the connection params (tag 0x84)")
+ ap.add_argument("--no-sec", action="store_true", help="omit the security params (tag 0x85)")
+ ap.add_argument("--no-http", action="store_true", help="omit the HTTP POST params (tag 0x89)")
+
+ args = ap.parse_args()
+
+ missing = []
+ if not args.no_conn:
+ missing += [n for n in ('ip', 'port', 'buffer') if getattr(args, n) is None]
+ if not args.no_sec:
+ missing += [n for n in ('psk_id', 'kvn', 'kid_ref') if getattr(args, n) is None]
+ if not args.no_http:
+ missing += [n for n in ('host', 'uri') if getattr(args, n) is None]
+ if missing:
+ ap.error("pass every value required: %s." % " ".join("--" + n.replace('_', '-') for n in missing))
+
+ session = []
+ if not args.no_conn:
+ session.append(RasConnectionParams(children=[
+ BearerDescription(decoded={'bearer_type': 'default', 'bearer_parameters': ''}),
+ BufferSize(decoded=args.buffer),
+ UiccTransportLevel(decoded={'protocol_type': 'tcp_uicc_client_remote',
+ 'port_number': args.port}),
+ OtherAddress(decoded={'type_of_address': 'ipv4',
+ 'address': bytes(int(b) for b in args.ip.split("."))})]))
+ if not args.no_sec:
+ session.append(SecurityParams(decoded={'psk_id': args.psk_id.encode(), 'kvn': args.kvn,
+ 'kid': args.kid_ref, 'sha_type': None}))
+ if not args.no_http:
+ session.append(HttpPostParams(children=[AdminHostParam(decoded=args.host),
+ AdminUriParam(decoded=args.uri)]))
+ trig = AdmSessTriggerParams(children=[AdmSessionParams(children=session)]).to_tlv()
+
+ # stderr for logs, stdout for data
+ print("# trigger TLV %d B %s" % (len(trig), trig.hex()), file=sys.stderr)
+ print("# %-13s %d B %s" % ("secured data", len(trig), trig.hex()), file=sys.stderr)
+ print(trig.hex())
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/43566?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: Ic646626319cd7a26195d0ffff55900032abcddeb
Gerrit-Change-Number: 43566
Gerrit-PatchSet: 1
Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>
1
0
[M] Change in pysim[master]: bip/smpp2sim: TERMINAL PROFILE that matches what we do
by Hoernchen 08 Sep '26
by Hoernchen 08 Sep '26
08 Sep '26
Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/43564?usp=email )
Change subject: bip/smpp2sim: TERMINAL PROFILE that matches what we do
......................................................................
bip/smpp2sim: TERMINAL PROFILE that matches what we do
pySim-smpp2sim sends "ff" * 32, that byte list tells every card the
terminal has a display, a keypad, a second card slot, a radio it can
query for location and NMR, five BIP bearers and six transport modes and
a toaster and a dog according to TS 102 223 5.2
We only have the twelfth byte and one bit of the seventeenth and no dog.
A card issues annoying weird things like PROVIDE LOCAL INFORMATION or
UDP only because the profile said so, so stop pretending we know what any
of that is.
Annex T table T.1 lists what a Connected Entity (a CAT client that is not
the modem) may announce. Announce that and the SMS-PP download and
SEND SHORT MESSAGE bits the OTA path needs, only the bearer is obviously
made up, it's the host TCP stack and 5.2 closest match is GPRS.
We can still extend and change all of this, but for now something that
works and constrains what the card asks for and is therefore actually
reproducible is important.
Change-Id: I91a60760fc3ad816b7385da8b26ec7330b462abd
---
M pySim-smpp2sim.py
M pySim/bip.py
M tests/unittests/test_bip_relay.py
3 files changed, 76 insertions(+), 5 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/64/43564/1
diff --git a/pySim-smpp2sim.py b/pySim-smpp2sim.py
index 95b9ccd..c1b2bb7 100755
--- a/pySim-smpp2sim.py
+++ b/pySim-smpp2sim.py
@@ -50,7 +50,7 @@
from pySim.sms import SMS_DELIVER, SMS_SUBMIT, AddressField
-from pySim.bip import Proact
+from pySim.bip import Proact, terminal_profile
from pySim.transport import LinkBase, ProactiveHandler, argparse_add_reader_args, init_reader, ApduTracer
from pySim.commands import SimCardCommands
from pySim.cards import UiccCardBase
@@ -126,8 +126,7 @@
self.scc.sel_ctrl = "0004"
self.card.read_aids()
self.card.select_adf_by_aid(adf='usim')
- # FIXME: create a more realistic profile than ffffff
- self.scc.terminal_profile('ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff')
+ self.scc.terminal_profile(b2h(terminal_profile()))
# Connect the BIP relay inbound path to the card.
# relay socket receives data -> ME initiated ENVELOPE EVENT DOWNLOA
# -> triggers RECEIVE DATA proactive session.
diff --git a/pySim/bip.py b/pySim/bip.py
index 8370b5b..b25fe5f 100644
--- a/pySim/bip.py
+++ b/pySim/bip.py
@@ -46,6 +46,35 @@
logger = logging.getLogger(__name__)
+
+def terminal_profile(num_channels: int = 7) -> bytes:
+ """TERMINAL PROFILE for what we implement, TS 102 223 5.2 and annex T.
+
+ Annex T table T.1 lists what a Connected Entity, a CAT client that is not the modem
+ which is pretty much what we are, may announce, and its inverse is what only a modem may announce.
+ """
+ if not 0 <= num_channels <= ProactChannels.MAX_CHANNELS:
+ raise ValueError('num_channels must be 0..%u' % ProactChannels.MAX_CHANNELS)
+ profile = bytearray(32)
+ # 1 (Download): b1 profile download, b2+b5 SMS-PP data download. Both of the latter, per the
+ # note in TS 31.111 5.2: "several bits may need to be set to 1 for the support of the same
+ # facility ... because of backward compatibility with SAT". The relay is OTA over SMS-PP.
+ profile[0] = 0x01 | 0x02 | 0x10
+ profile[1] = 0x01 # 2 (Other): b1 command result
+ profile[2] = 0x80 # 3: b8 REFRESH (empty result is a valid answer, 6.4.7)
+ profile[3] = 0x02 # 4: b2 SEND SHORT MESSAGE (the OTA response path)
+ profile[4] = 0x01 # 5: b1 SET UP EVENT LIST
+ profile[5] = 0x04 | 0x08 # 6: b3 Event Data available, b4 Event Channel status
+ # 12 (class "e"): b1..b5 OPEN CHANNEL, CLOSE CHANNEL, RECEIVE DATA, SEND DATA, GET CHANNEL
+ # STATUS.
+ profile[11] = 0x1f
+ # 13 (class "e" supported bearers): b2 GPRS, and b6..b8 the number of channels.
+ profile[12] = 0x02 | (num_channels << 5)
+ profile[13] = 0x40 | 0x20 # 14: b6 no display capability, b7 no keypad available
+ profile[16] = 0x01 # 15: b1 TCP, UICC in client mode, remote connection
+ return bytes(profile)
+
+
class ProactChannel:
"""Representation of a single BIP channel, backed by a blocking TCP
socket.
@@ -168,6 +197,11 @@
class ProactChannels:
"""Wrapper class for maintaining state of proactive channels."""
+
+ # TS 102 223 8.56 channel identifier in 3 bits as "1 to 7", 0 == no channel available
+ # TERMINAL PROFILE has to agree with byte 13 , "number of channels supported by terminal"
+ MAX_CHANNELS = 7
+
def __init__(self, on_data_available=None):
self.channels = {}
# called from a channel rx reader thread on empty->non-empty buf
@@ -176,7 +210,7 @@
def channel_create(self) -> ProactChannel:
"""Create a new proactive channel, allocating its integer number."""
- for i in range(1, 8):
+ for i in range(1, self.MAX_CHANNELS + 1):
if not i in self.channels:
self.channels[i] = ProactChannel(self, i)
return self.channels[i]
diff --git a/tests/unittests/test_bip_relay.py b/tests/unittests/test_bip_relay.py
index 8716585..e7229d7 100644
--- a/tests/unittests/test_bip_relay.py
+++ b/tests/unittests/test_bip_relay.py
@@ -27,7 +27,7 @@
OtherAddress, ChannelData, ChannelDataLength, ChannelStatus,
Result)
-from pySim.bip import Proact
+from pySim.bip import Proact, ProactChannels, terminal_profile
class _EchoServer:
@@ -324,6 +324,44 @@
self.assertEqual(self.proact.channels.channels, {}) # channel given back
+class TerminalProfileTest(unittest.TestCase):
+ """TS 102 223 5.2, one bit per CAT facility"""
+
+ def setUp(self):
+ self.profile = terminal_profile()
+
+ def byte(self, n):
+ return self.profile[n - 1] # 1-based, as 5.2 numbers them
+
+ def test_announced(self):
+ self.assertEqual(len(self.profile), 32)
+ self.assertEqual(self.byte(1), 0x13) # profile download, SMS-PP download b2+b5
+ self.assertEqual(self.byte(4), 0x02) # SEND SHORT MESSAGE
+ self.assertEqual(self.byte(5) & 0x01, 0x01) # SET UP EVENT LIST
+ self.assertEqual(self.byte(6), 0x0c) # events: data available, channel status
+ self.assertEqual(self.byte(12), 0x1f) # OPEN/CLOSE CHANNEL, RECEIVE/SEND DATA, STATUS
+ self.assertEqual(self.byte(13) >> 5, ProactChannels.MAX_CHANNELS)
+ self.assertEqual(self.byte(14), 0x60) # class ND, class NK
+ self.assertEqual(self.byte(17), 0x01) # TCP, UICC client mode, remote
+
+ def test_not_announced(self):
+ self.assertEqual(self.byte(3) & 0x60, 0) # POLL INTERVAL, POLLING OFF
+ self.assertEqual(self.byte(4) & 0xc0, 0) # PROVIDE LOCAL INFORMATION, NMR
+ self.assertEqual(self.byte(12) & 0xe0, 0) # SERVICE SEARCH/INFORMATION, DECLARE SERVICE
+ self.assertEqual(self.byte(14) & 0x1f, 0) # no characters down the display
+ for n in (7, 9, 10, 11, 15, 16, 18): # class "a", class "d", display, ESN/IMEISV
+ self.assertEqual(self.byte(n), 0)
+
+ def test_channel_count(self):
+ self.assertEqual(terminal_profile(3)[12] >> 5, 3)
+ with self.assertRaises(ValueError): # 8.56: 1 to 7
+ terminal_profile(8)
+
+
+if __name__ == "__main__":
+ unittest.main()
+
+
class BipSinkTest(unittest.TestCase):
"""Both sinks are optional, a driver with no SMS path at all must not crash and burn
with a card that sends one, and one that has one must get the PDU."""
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/43564?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: I91a60760fc3ad816b7385da8b26ec7330b462abd
Gerrit-Change-Number: 43564
Gerrit-PatchSet: 1
Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>
1
0
[S] Change in pysim[master]: tests: stop test_log from leaking the print callback
by Hoernchen 08 Sep '26
by Hoernchen 08 Sep '26
08 Sep '26
Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/43539?usp=email )
Change subject: tests: stop test_log from leaking the print callback
......................................................................
tests: stop test_log from leaking the print callback
PySimLogger.setup() installs a process-global print callback.
PySimLogger_Test sets one, a helper that asserts the message equals a global
expected_message, and never removes it, so from the moment test_log runs,
every PySimLogger message emitted anywhere in the process is checked against
whatever string that global happens to hold.
Fortunately unittest discovery runs modules in sorted order, and today
the PySimLogger users that log during tests all sort before test_log, so
this only breaks as soon as I try to add tests, just like anything else
breaks as soon as I try to use it.
Change-Id: I481e2c443fe0f412380b0f1acf6da5971ffca147
---
M tests/unittests/test_log.py
1 file changed, 11 insertions(+), 0 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/39/43539/1
diff --git a/tests/unittests/test_log.py b/tests/unittests/test_log.py
index a8e38dd..ac651ae 100755
--- a/tests/unittests/test_log.py
+++ b/tests/unittests/test_log.py
@@ -37,6 +37,17 @@
class PySimLogger_Test(unittest.TestCase):
+ def setUp(self):
+ # PySimLogger.setup() is global, so a print callback left installed here fires for
+ # every PySimLogger message emitted by any test module that runs later in the same process
+ # ... where it asserts against a stale 'expected_message' and fails a test that has nothing
+ # to do with logging. Great fun!
+ # Restore before each test.
+ saved = (PySimLogger.print_callback, PySimLogger.verbose)
+ def _restore():
+ PySimLogger.print_callback, PySimLogger.verbose = saved
+ self.addCleanup(_restore)
+
def __test_01_safe_defaults_one(self, callback, message:str):
# When log messages are sent to an unconfigured PySimLogger class, we expect the unmodified message being
# logged to stdout, just as if it were printed via a normal print() statement.
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/43539?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: I481e2c443fe0f412380b0f1acf6da5971ffca147
Gerrit-Change-Number: 43539
Gerrit-PatchSet: 1
Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>
1
0
08 Sep '26
Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/43540?usp=email )
Change subject: bip: move the BIP relay into pySim.bip
......................................................................
bip: move the BIP relay into pySim.bip
Move the code from pySim-smpp2sim.py to its own file, so it can be properly
extended.
The current file parses argv, opens a reader and starts the Twisted
reactor at import time, so nothing else can import it.
No functional changes yet, improvements follow in later commits.
Change-Id: Ifd8a15684939977d29ea83a6b669daee14484e88
---
M pySim-smpp2sim.py
A pySim/bip.py
A tests/unittests/test_bip_relay.py
3 files changed, 296 insertions(+), 207 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/40/43540/1
diff --git a/pySim-smpp2sim.py b/pySim-smpp2sim.py
index fa670e5..2c7ff51 100755
--- a/pySim-smpp2sim.py
+++ b/pySim-smpp2sim.py
@@ -47,6 +47,7 @@
from pySim.sms import SMS_DELIVER, SMS_SUBMIT, AddressField
+from pySim.bip import Proact
from pySim.transport import LinkBase, ProactiveHandler, argparse_add_reader_args, init_reader, ApduTracer
from pySim.commands import SimCardCommands
from pySim.cards import UiccCardBase
@@ -83,212 +84,6 @@
"""called by twisted TCP client."""
logger.error("%s: connected!" % p)
-class ProactChannel:
- """Representation of a single protective channel."""
- def __init__(self, channels: 'ProactChannels', chan_nr: int):
- self.channels = channels
- self.chan_nr = chan_nr
- self.ep = None
-
- def close(self):
- """Close the channel."""
- if self.ep:
- self.ep.disconnect()
- self.channels.channel_delete(self.chan_nr)
-
-class ProactChannels:
- """Wrapper class for maintaining state of proactive channels."""
- def __init__(self):
- self.channels = {}
-
- def channel_create(self) -> ProactChannel:
- """Create a new proactive channel, allocating its integer number."""
- for i in range(1, 9):
- if not i in self.channels:
- self.channels[i] = ProactChannel(self, i)
- return self.channels[i]
- raise ValueError('Cannot allocate another channel: All channels active')
-
- def channel_delete(self, chan_nr: int):
- del self.channels[chan_nr]
-
-class Proact(ProactiveHandler):
- #def __init__(self, smpp_factory):
- # self.smpp_factory = smpp_factory
- def __init__(self):
- self.channels = ProactChannels()
-
- @staticmethod
- def _find_first_element_of_type(instlist, cls):
- for i in instlist:
- if isinstance(i, cls):
- return i
- return None
-
- """Call-back which the pySim transport core calls whenever it receives a
- proactive command from the SIM."""
- def handle_SendShortMessage(self, pcmd: ProactiveCommand):
- # {'smspp_download': [{'device_identities': {'source_dev_id': 'network',
- # 'dest_dev_id': 'uicc'}},
- # {'address': {'ton_npi': {'ext': True,
- # 'type_of_number': 'international',
- # 'numbering_plan_id': 'isdn_e164'},
- # 'call_number': '79'}},
- # {'sms_tpdu': {'tpdu': '40048111227ff6407070611535004d02700000481516011212000001fe4c0943aea42e45021c078ae06c66afc09303608874b72f58bacadb0dcf665c29349c799fbb522e61709c9baf1890015e8e8e196e36153106c8b92f95153774'}}
- # ]}
- """Card requests sending a SMS. We need to pass it on to the ESME via SMPP."""
- logger.info("SendShortMessage")
- logger.info(pcmd)
- # Relevant parts in pcmd: Address, SMS_TPDU
- addr_ie = Proact._find_first_element_of_type(pcmd.children, Address)
- sms_tpdu_ie = Proact._find_first_element_of_type(pcmd.children, SMS_TPDU)
- raw_tpdu = sms_tpdu_ie.decoded['tpdu']
- submit = SMS_SUBMIT.from_bytes(raw_tpdu)
- submit.tp_da = AddressField(addr_ie.decoded['call_number'], addr_ie.decoded['ton_npi']['type_of_number'],
- addr_ie.decoded['ton_npi']['numbering_plan_id'])
- logger.info(submit)
- self.send_sms_via_smpp(submit)
-
- def handle_OpenChannel(self, pcmd: ProactiveCommand):
- """Card requests opening a new channel via a UDP/TCP socket."""
- # {'open_channel': [{'command_details': {'command_number': 1,
- # 'type_of_command': 'open_channel',
- # 'command_qualifier': 3}},
- # {'device_identities': {'source_dev_id': 'uicc',
- # 'dest_dev_id': 'terminal'}},
- # {'bearer_description': {'bearer_type': 'default',
- # 'bearer_parameters': ''}},
- # {'buffer_size': 1024},
- # {'uicc_transport_level': {'protocol_type': 'tcp_uicc_client_remote',
- # 'port_number': 32768}},
- # {'other_address': {'type_of_address': 'ipv4',
- # 'address': '01020304'}}
- # ]}
- logger.info("OpenChannel")
- logger.info(pcmd)
- transp_lvl_ie = Proact._find_first_element_of_type(pcmd.children, UiccTransportLevel)
- other_addr_ie = Proact._find_first_element_of_type(pcmd.children, OtherAddress)
- bearer_desc_ie = Proact._find_first_element_of_type(pcmd.children, BearerDescription)
- buffer_size_ie = Proact._find_first_element_of_type(pcmd.children, BufferSize)
- if transp_lvl_ie.decoded['protocol_type'] != 'tcp_uicc_client_remote':
- raise ValueError('Unsupported protocol_type')
- if other_addr_ie.decoded.get('type_of_address', None) != 'ipv4':
- raise ValueError('Unsupported type_of_address')
- ipv4_bytes = h2b(other_addr_ie.decoded['address'])
- ipv4_str = '%u.%u.%u.%u' % (ipv4_bytes[0], ipv4_bytes[1], ipv4_bytes[2], ipv4_bytes[3])
- port_nr = transp_lvl_ie.decoded['port_number']
- print("%s:%u" % (ipv4_str, port_nr))
- channel = self.channels.channel_create()
- channel.ep = endpoints.TCP4ClientEndpoint(reactor, ipv4_str, port_nr)
- channel.prot = TcpProtocol()
- d = endpoints.connectProtocol(channel.ep, channel.prot)
- # FIXME: why is this never called despite the client showing the inbound connection?
- d.addCallback(tcp_connected_callback)
-
- # Terminal Response example: [
- # {'command_details': {'command_number': 1,
- # 'type_of_command': 'open_channel',
- # 'command_qualifier': 3}},
- # {'device_identities': {'source_dev_id': 'terminal', 'dest_dev_id': 'uicc'}},
- # {'result': {'general_result': 'performed_successfully', 'additional_information': ''}},
- # {'channel_status': '8100'},
- # {'bearer_description': {'bearer_type': 'default', 'bearer_parameters': ''}},
- # {'buffer_size': 1024}
- # ]
- return self.prepare_response(pcmd) + [ChannelStatus(decoded='8100'), bearer_desc_ie, buffer_size_ie]
-
- def handle_CloseChannel(self, pcmd: ProactiveCommand):
- """Close a channel."""
- logger.info("CloseChannel")
- logger.info(pcmd)
-
- def handle_ReceiveData(self, pcmd: ProactiveCommand):
- """Receive/read data from the socket."""
- # {'receive_data': [{'command_details': {'command_number': 1,
- # 'type_of_command': 'receive_data',
- # 'command_qualifier': 0}},
- # {'device_identities': {'source_dev_id': 'uicc',
- # 'dest_dev_id': 'channel_1'}},
- # {'channel_data_length': 9}
- # ]}
- logger.info("ReceiveData")
- logger.info(pcmd)
- # Terminal Response example: [
- # {'command_details': {'command_number': 1,
- # 'type_of_command': 'receive_data',
- # 'command_qualifier': 0}},
- # {'device_identities': {'source_dev_id': 'terminal', 'dest_dev_id': 'uicc'}},
- # {'result': {'general_result': 'performed_successfully', 'additional_information': ''}},
- # {'channel_data': '16030100040e000000'},
- # {'channel_data_length': 0}
- # ]
- return self.prepare_response(pcmd) + []
-
- def handle_SendData(self, pcmd: ProactiveCommand):
- """Send/write data received from the SIM to the socket."""
- # {'send_data': [{'command_details': {'command_number': 1,
- # 'type_of_command': 'send_data',
- # 'command_qualifier': 1}},
- # {'device_identities': {'source_dev_id': 'uicc',
- # 'dest_dev_id': 'channel_1'}},
- # {'channel_data': '160301003c010000380303d0f45e12b52ce5bb522750dd037738195334c87a46a847fe2b6886cada9ea6bf00000a00ae008c008b00b0002c010000050001000101'}
- # ]}
- logger.info("SendData")
- logger.info(pcmd)
- dev_id_ie = Proact._find_first_element_of_type(pcmd.children, DeviceIdentities)
- chan_data_ie = Proact._find_first_element_of_type(pcmd.children, ChannelData)
- chan_str = dev_id_ie.decoded['dest_dev_id']
- chan_nr = 1 # FIXME
- chan = self.channels.channels.get(chan_nr, None)
- # FIXME chan.prot.transport.write(h2b(chan_data_ie.decoded))
- # Terminal Response example: [
- # {'command_details': {'command_number': 1,
- # 'type_of_command': 'send_data',
- # 'command_qualifier': 1}},
- # {'device_identities': {'source_dev_id': 'terminal', 'dest_dev_id': 'uicc'}},
- # {'result': {'general_result': 'performed_successfully', 'additional_information': ''}},
- # {'channel_data_length': 255}
- # ]
- return self.prepare_response(pcmd) + [ChannelDataLength(decoded=255)]
-
- def handle_SetUpEventList(self, pcmd: ProactiveCommand):
- # {'set_up_event_list': [{'command_details': {'command_number': 1,
- # 'type_of_command': 'set_up_event_list',
- # 'command_qualifier': 0}},
- # {'device_identities': {'source_dev_id': 'uicc',
- # 'dest_dev_id': 'terminal'}},
- # {'event_list': ['data_available', 'channel_status']}
- # ]}
- logger.info("SetUpEventList")
- logger.info(pcmd)
- # Terminal Response example: [
- # {'command_details': {'command_number': 1,
- # 'type_of_command': 'set_up_event_list',
- # 'command_qualifier': 0}},
- # {'device_identities': {'source_dev_id': 'terminal', 'dest_dev_id': 'uicc'}},
- # {'result': {'general_result': 'performed_successfully', 'additional_information': ''}}
- # ]
- return self.prepare_response(pcmd)
-
- def getChannelStatus(self, pcmd: ProactiveCommand):
- logger.info("GetChannelStatus")
- logger.info(pcmd)
- return self.prepare_response(pcmd) + []
-
- def send_sms_via_smpp(self, submit: SMS_SUBMIT):
- # while in a normal network the phone/ME would *submit* a message to the SMSC,
- # we are actually emulating the SMSC itself, so we must *deliver* the message
- # to the ESME
- deliver = SMS_DELIVER.from_submit(submit)
- deliver_smpp = deliver.to_smpp()
-
- hackish_global_smpp.sendDataRequest(deliver_smpp)
-# # obtain the connection/binding of system_id to be used for delivering MO-SMS to the ESME
-# connection = smpp_server.getBoundConnections[system_id].getNextBindingForDelivery()
-# connection.sendDataRequest(deliver_smpp)
-
-
-
def dcs_is_8bit(dcs):
if dcs == pdu_types.DataCoding(pdu_types.DataCodingScheme.DEFAULT,
pdu_types.DataCodingDefault.OCTET_UNSPECIFIED):
@@ -416,7 +211,8 @@
opts = option_parser.parse_args()
- tp = init_reader(opts, proactive_handler = Proact())
+ tp = init_reader(opts, proactive_handler = Proact(
+ sms_sink=lambda pdu: hackish_global_smpp.sendDataRequest(pdu)))
if tp is None:
exit(1)
tp.connect()
diff --git a/pySim/bip.py b/pySim/bip.py
new file mode 100644
index 0000000..287f31d
--- /dev/null
+++ b/pySim/bip.py
@@ -0,0 +1,248 @@
+# -*- coding: utf-8 -*-
+"""Bearer Independent Protocol relay"""
+
+#
+# (C) 2023-2024 by Harald Welte <laforge(a)osmocom.org>
+#
+# 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 logging
+import socket
+import threading
+import time
+
+from osmocom.utils import b2h, h2b
+
+from pySim.transport import ProactiveHandler
+from pySim.sms import SMS_DELIVER, SMS_SUBMIT, AddressField
+from pySim.cat import (ProactiveCommand, SendShortMessage, SMS_TPDU, SMSPPDownload,
+ BearerDescription, DeviceIdentities, Address, OtherAddress,
+ UiccTransportLevel, BufferSize, ChannelStatus, ChannelData,
+ ChannelDataLength, EventList, EventDownload, Result)
+
+logger = logging.getLogger(__name__)
+
+class ProactChannel:
+ """Representation of a single protective channel."""
+ def __init__(self, channels: 'ProactChannels', chan_nr: int):
+ self.channels = channels
+ self.chan_nr = chan_nr
+ self.ep = None
+
+ def close(self):
+ """Close the channel."""
+ if self.ep:
+ self.ep.disconnect()
+ self.channels.channel_delete(self.chan_nr)
+
+class ProactChannels:
+ """Wrapper class for maintaining state of proactive channels."""
+ def __init__(self):
+ self.channels = {}
+
+ def channel_create(self) -> ProactChannel:
+ """Create a new proactive channel, allocating its integer number."""
+ for i in range(1, 9):
+ if not i in self.channels:
+ self.channels[i] = ProactChannel(self, i)
+ return self.channels[i]
+ raise ValueError('Cannot allocate another channel: All channels active')
+
+ def channel_delete(self, chan_nr: int):
+ del self.channels[chan_nr]
+
+class Proact(ProactiveHandler):
+ #def __init__(self, smpp_factory):
+ # self.smpp_factory = smpp_factory
+ def __init__(self, sms_sink=None):
+ # sms_sink(pdu) delivers an MO-SMS the card sent (SEND SHORT MESSAGE)
+ # onwards; pySim-smpp2sim.py hands it to its SMPP server. None -> log
+ # and drop.
+ self.sms_sink = sms_sink
+ self.channels = ProactChannels()
+
+ @staticmethod
+ def _find_first_element_of_type(instlist, cls):
+ for i in instlist:
+ if isinstance(i, cls):
+ return i
+ return None
+
+ """Call-back which the pySim transport core calls whenever it receives a
+ proactive command from the SIM."""
+ def handle_SendShortMessage(self, pcmd: ProactiveCommand):
+ # {'smspp_download': [{'device_identities': {'source_dev_id': 'network',
+ # 'dest_dev_id': 'uicc'}},
+ # {'address': {'ton_npi': {'ext': True,
+ # 'type_of_number': 'international',
+ # 'numbering_plan_id': 'isdn_e164'},
+ # 'call_number': '79'}},
+ # {'sms_tpdu': {'tpdu': '40048111227ff6407070611535004d02700000481516011212000001fe4c0943aea42e45021c078ae06c66afc09303608874b72f58bacadb0dcf665c29349c799fbb522e61709c9baf1890015e8e8e196e36153106c8b92f95153774'}}
+ # ]}
+ """Card requests sending a SMS. We need to pass it on to the ESME via SMPP."""
+ logger.info("SendShortMessage")
+ logger.info(pcmd)
+ # Relevant parts in pcmd: Address, SMS_TPDU
+ addr_ie = Proact._find_first_element_of_type(pcmd.children, Address)
+ sms_tpdu_ie = Proact._find_first_element_of_type(pcmd.children, SMS_TPDU)
+ raw_tpdu = sms_tpdu_ie.decoded['tpdu']
+ submit = SMS_SUBMIT.from_bytes(raw_tpdu)
+ submit.tp_da = AddressField(addr_ie.decoded['call_number'], addr_ie.decoded['ton_npi']['type_of_number'],
+ addr_ie.decoded['ton_npi']['numbering_plan_id'])
+ logger.info(submit)
+ self.send_sms_via_smpp(submit)
+
+ def handle_OpenChannel(self, pcmd: ProactiveCommand):
+ """Card requests opening a new channel via a UDP/TCP socket."""
+ # {'open_channel': [{'command_details': {'command_number': 1,
+ # 'type_of_command': 'open_channel',
+ # 'command_qualifier': 3}},
+ # {'device_identities': {'source_dev_id': 'uicc',
+ # 'dest_dev_id': 'terminal'}},
+ # {'bearer_description': {'bearer_type': 'default',
+ # 'bearer_parameters': ''}},
+ # {'buffer_size': 1024},
+ # {'uicc_transport_level': {'protocol_type': 'tcp_uicc_client_remote',
+ # 'port_number': 32768}},
+ # {'other_address': {'type_of_address': 'ipv4',
+ # 'address': '01020304'}}
+ # ]}
+ logger.info("OpenChannel")
+ logger.info(pcmd)
+ transp_lvl_ie = Proact._find_first_element_of_type(pcmd.children, UiccTransportLevel)
+ other_addr_ie = Proact._find_first_element_of_type(pcmd.children, OtherAddress)
+ bearer_desc_ie = Proact._find_first_element_of_type(pcmd.children, BearerDescription)
+ buffer_size_ie = Proact._find_first_element_of_type(pcmd.children, BufferSize)
+ if transp_lvl_ie.decoded['protocol_type'] != 'tcp_uicc_client_remote':
+ raise ValueError('Unsupported protocol_type')
+ if other_addr_ie.decoded.get('type_of_address', None) != 'ipv4':
+ raise ValueError('Unsupported type_of_address')
+ ipv4_bytes = h2b(other_addr_ie.decoded['address'])
+ ipv4_str = '%u.%u.%u.%u' % (ipv4_bytes[0], ipv4_bytes[1], ipv4_bytes[2], ipv4_bytes[3])
+ port_nr = transp_lvl_ie.decoded['port_number']
+ print("%s:%u" % (ipv4_str, port_nr))
+ channel = self.channels.channel_create()
+ channel.ep = endpoints.TCP4ClientEndpoint(reactor, ipv4_str, port_nr)
+ channel.prot = TcpProtocol()
+ d = endpoints.connectProtocol(channel.ep, channel.prot)
+ # FIXME: why is this never called despite the client showing the inbound connection?
+ d.addCallback(tcp_connected_callback)
+
+ # Terminal Response example: [
+ # {'command_details': {'command_number': 1,
+ # 'type_of_command': 'open_channel',
+ # 'command_qualifier': 3}},
+ # {'device_identities': {'source_dev_id': 'terminal', 'dest_dev_id': 'uicc'}},
+ # {'result': {'general_result': 'performed_successfully', 'additional_information': ''}},
+ # {'channel_status': '8100'},
+ # {'bearer_description': {'bearer_type': 'default', 'bearer_parameters': ''}},
+ # {'buffer_size': 1024}
+ # ]
+ return self.prepare_response(pcmd) + [ChannelStatus(decoded='8100'), bearer_desc_ie, buffer_size_ie]
+
+ def handle_CloseChannel(self, pcmd: ProactiveCommand):
+ """Close a channel."""
+ logger.info("CloseChannel")
+ logger.info(pcmd)
+
+ def handle_ReceiveData(self, pcmd: ProactiveCommand):
+ """Receive/read data from the socket."""
+ # {'receive_data': [{'command_details': {'command_number': 1,
+ # 'type_of_command': 'receive_data',
+ # 'command_qualifier': 0}},
+ # {'device_identities': {'source_dev_id': 'uicc',
+ # 'dest_dev_id': 'channel_1'}},
+ # {'channel_data_length': 9}
+ # ]}
+ logger.info("ReceiveData")
+ logger.info(pcmd)
+ # Terminal Response example: [
+ # {'command_details': {'command_number': 1,
+ # 'type_of_command': 'receive_data',
+ # 'command_qualifier': 0}},
+ # {'device_identities': {'source_dev_id': 'terminal', 'dest_dev_id': 'uicc'}},
+ # {'result': {'general_result': 'performed_successfully', 'additional_information': ''}},
+ # {'channel_data': '16030100040e000000'},
+ # {'channel_data_length': 0}
+ # ]
+ return self.prepare_response(pcmd) + []
+
+ def handle_SendData(self, pcmd: ProactiveCommand):
+ """Send/write data received from the SIM to the socket."""
+ # {'send_data': [{'command_details': {'command_number': 1,
+ # 'type_of_command': 'send_data',
+ # 'command_qualifier': 1}},
+ # {'device_identities': {'source_dev_id': 'uicc',
+ # 'dest_dev_id': 'channel_1'}},
+ # {'channel_data': '160301003c010000380303d0f45e12b52ce5bb522750dd037738195334c87a46a847fe2b6886cada9ea6bf00000a00ae008c008b00b0002c010000050001000101'}
+ # ]}
+ logger.info("SendData")
+ logger.info(pcmd)
+ dev_id_ie = Proact._find_first_element_of_type(pcmd.children, DeviceIdentities)
+ chan_data_ie = Proact._find_first_element_of_type(pcmd.children, ChannelData)
+ chan_str = dev_id_ie.decoded['dest_dev_id']
+ chan_nr = 1 # FIXME
+ chan = self.channels.channels.get(chan_nr, None)
+ # FIXME chan.prot.transport.write(h2b(chan_data_ie.decoded))
+ # Terminal Response example: [
+ # {'command_details': {'command_number': 1,
+ # 'type_of_command': 'send_data',
+ # 'command_qualifier': 1}},
+ # {'device_identities': {'source_dev_id': 'terminal', 'dest_dev_id': 'uicc'}},
+ # {'result': {'general_result': 'performed_successfully', 'additional_information': ''}},
+ # {'channel_data_length': 255}
+ # ]
+ return self.prepare_response(pcmd) + [ChannelDataLength(decoded=255)]
+
+ def handle_SetUpEventList(self, pcmd: ProactiveCommand):
+ # {'set_up_event_list': [{'command_details': {'command_number': 1,
+ # 'type_of_command': 'set_up_event_list',
+ # 'command_qualifier': 0}},
+ # {'device_identities': {'source_dev_id': 'uicc',
+ # 'dest_dev_id': 'terminal'}},
+ # {'event_list': ['data_available', 'channel_status']}
+ # ]}
+ logger.info("SetUpEventList")
+ logger.info(pcmd)
+ # Terminal Response example: [
+ # {'command_details': {'command_number': 1,
+ # 'type_of_command': 'set_up_event_list',
+ # 'command_qualifier': 0}},
+ # {'device_identities': {'source_dev_id': 'terminal', 'dest_dev_id': 'uicc'}},
+ # {'result': {'general_result': 'performed_successfully', 'additional_information': ''}}
+ # ]
+ return self.prepare_response(pcmd)
+
+ def getChannelStatus(self, pcmd: ProactiveCommand):
+ logger.info("GetChannelStatus")
+ logger.info(pcmd)
+ return self.prepare_response(pcmd) + []
+
+ def send_sms_via_smpp(self, submit: SMS_SUBMIT):
+ # while in a normal network the phone/ME would *submit* a message to the SMSC,
+ # we are actually emulating the SMSC itself, so we must *deliver* the message
+ # to the ESME
+ deliver = SMS_DELIVER.from_submit(submit)
+ deliver_smpp = deliver.to_smpp()
+
+ if self.sms_sink is None:
+ logger.info('no sms_sink: dropping MO-SMS %s', deliver_smpp)
+ return
+ self.sms_sink(deliver_smpp)
+# # obtain the connection/binding of system_id to be used for delivering MO-SMS to the ESME
+# connection = smpp_server.getBoundConnections[system_id].getNextBindingForDelivery()
+# connection.sendDataRequest(deliver_smpp)
+
+
+
diff --git a/tests/unittests/test_bip_relay.py b/tests/unittests/test_bip_relay.py
new file mode 100644
index 0000000..e2e7ff8
--- /dev/null
+++ b/tests/unittests/test_bip_relay.py
@@ -0,0 +1,45 @@
+#!/usr/bin/env python3
+
+# (C) 2026 by sysmocom - s.f.m.c. GmbH <info(a)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 unittest
+
+from pySim.bip import Proact
+from pySim.sms import SMS_SUBMIT, AddressField
+
+
+class BipSinkTest(unittest.TestCase):
+ """Both sinks are optional, a driver with no SMS path at all must not crash and burn
+ with a card that sends one, and one that has one must get the PDU."""
+
+ def _submit(self):
+ return SMS_SUBMIT(tp_da=AddressField('12345', 'unknown', 'unknown'),
+ tp_ud=b'\x01\x02', tp_udl=2, tp_dcs=0xf6)
+
+ def test_sinks_default_to_none(self):
+ p = Proact()
+ self.assertIsNone(p.sms_sink)
+
+ def test_mo_sms_goes_to_the_sink(self):
+ seen = []
+ Proact(sms_sink=seen.append).send_sms_via_smpp(self._submit())
+ self.assertEqual(len(seen), 1)
+
+ def test_no_sms_sink_drops_instead_of_raising(self):
+ with self.assertLogs('pySim.bip', level='INFO'):
+ Proact().send_sms_via_smpp(self._submit())
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/43540?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: Ifd8a15684939977d29ea83a6b669daee14484e88
Gerrit-Change-Number: 43540
Gerrit-PatchSet: 1
Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>
1
0
[L] Change in pysim[master]: ota: add TS 102 226 5.2 Expanded Remote Application data format
by Hoernchen 08 Sep '26
by Hoernchen 08 Sep '26
08 Sep '26
Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/43541?usp=email )
Change subject: ota: add TS 102 226 5.2 Expanded Remote Application data format
......................................................................
ota: add TS 102 226 5.2 Expanded Remote Application data format
TS 102 226 section 5.2 "Command and Response Scripting templates"
The advantage over compact RFM/RAM commands is one C-APDU TLV per command
and one R-APDU TLV per result, so multiple commands return the response
of each one rather than only that of the last.
encode_expanded_cmd() builds the Command Scripting template
decode_expanded_resp() decodes the Response Scripting template into a
Container.
The 'truncated' key must be checked!
OtaDialect.encode_cmd()/decode_resp() now has a remote_format param for
compact and expanded formats, default stays compact,so existing code is
unaffected.
Change-Id: Idec00d16fd1a7d4a7129b2a3b6f0ef37dabcecb7
---
M pySim/ota.py
M tests/unittests/test_ota.py
2 files changed, 411 insertions(+), 9 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/41/43541/1
diff --git a/pySim/ota.py b/pySim/ota.py
index 3d2da15..30d9869 100644
--- a/pySim/ota.py
+++ b/pySim/ota.py
@@ -18,10 +18,12 @@
import zlib
import abc
import struct
-from typing import Optional, Tuple
+from typing import Optional, Tuple, List, Union
from construct import Enum, Int8ub, Int16ub, Struct, BitsInteger, BitStruct
from construct import Flag, Padding, Switch, this, PrefixedArray, GreedyRange
+from construct import Const, Prefixed, Select, Construct, SizeofError, stream_read, stream_write
from osmocom.construct import *
+from osmocom.tlv import bertlv_encode_len
from osmocom.utils import b2h
from pySim.sms import UserDataHeader
@@ -56,6 +58,173 @@
'last_status_word'/HexAdapter(Bytes(2)),
'last_response_data'/HexAdapter(GreedyBytes))
+######################################################################
+# Expanded Remote Application data format, ETSI TS 102 226 V19.0.0 (2025-11) Section 5.2
+# 5.2.1 Expanded Remote command structure
+# 5.2.1.1 C-APDU TLV
+# 5.2.1.2 Immediate Action TLV
+# 5.2.1.3 Error Action TLV
+# 5.2.1.4 Script Chaining TLV
+# 5.2.2 Expanded Remote response structure (tables 5.10 .. 5.16)
+#
+# BER-TLV tag values from ETSI TS 101 220 V19.0.0 tables 7.18, 7.19, 7.20
+# C-APDU / R-APDU ETSI TS 102 223 Section 8.35 + 8.36
+# inside these the CR flag of the tag is 0 (TS 101 220 tables 7.19/7.20),
+# so tag bytes are 22 and 23 and not A2/A3.
+#
+# This layer sits above the TS 102 225 security layer.
+######################################################################
+
+class BerTlvLength(Construct):
+ """A definite-length BER-TLV length field used by the "expanded remote
+ application data format" from ISO/IEC 8825-1 referenced by TS 102 226 5.2
+
+ - short form (0..127 -> single octet)
+ - long form (128.. -> 0x8N followed by N length octets)
+ Indefinite length coding (first octet 0x80, TS 102 226 tables 5.2a/5.10a)
+ is omitted here because it is only recommended for HTTPS/CoAP transport, not SMS."""
+ def _parse(self, stream, context, path):
+ first = stream_read(stream, 1, path)[0]
+ if first < 0x80:
+ return first
+ num_octets = first & 0x7f
+ if num_octets == 0:
+ raise NotImplementedError('indefinite coding is not supported')
+ return int.from_bytes(stream_read(stream, num_octets, path), 'big')
+
+ def _build(self, obj, stream, context, path):
+ encoded = bertlv_encode_len(obj)
+ stream_write(stream, encoded, len(encoded), path)
+ return obj
+
+ def _sizeof(self, context, path):
+ raise SizeofError('BER-TLV length has a variable size?!')
+
+BerTlvLen = BerTlvLength()
+
+class _RApduValueAdapter(Adapter):
+ """Split/join value of R-APDU COMPREHENSION-TLV TS 102 223 8.36
+ [R-APDU data (x-2 bytes)] SW1 SW2."""
+ def _decode(self, obj, context, path):
+ raw = bytes(obj)
+ return Container(response_data=b2h(raw[:-2]), status_word=b2h(raw[-2:]))
+
+ def _encode(self, obj, context, path):
+ return h2b(obj['response_data']) + h2b(obj['status_word'])
+
+#### Command Scripting template TS 102 226 table 5.2, TS 101 220 tables 7.18/7.19
+
+# TS 102 223 8.35
+ExpandedC_APDU = Struct('_tag'/Const(b'\x22'),
+ 'c_apdu'/Prefixed(BerTlvLen, HexAdapter(GreedyBytes)))
+
+ExpandedCmd = Struct('_tag'/Const(b'\xaa'),
+ 'commands'/Prefixed(BerTlvLen, GreedyRange(ExpandedC_APDU)))
+
+#### Response Scripting template TS 102 226 tables 5.10-5.16, TS 101 220 table 7.20
+
+# TS 102 223 8.36
+ExpandedR_APDU = Struct('_tag'/Const(b'\x23'),
+ 'r_apdu'/Prefixed(BerTlvLen, _RApduValueAdapter(GreedyBytes)))
+
+# TS 102 226 table 5.11
+# Value is an integer per ISO/IEC 8825-1, likely just one octet.
+ExpandedNumExecuted = Struct('_tag'/Const(b'\x80'),
+ 'number_of_commands'/Prefixed(BerTlvLen, GreedyInteger()))
+
+# TS 102 226 table 5.12
+ExpandedBadFormat = Struct('_tag'/Const(b'\x90'),
+ 'bad_format'/Prefixed(BerTlvLen,
+ Enum(Int8ub, unknown_tag=1, wrong_length=2, length_not_found=3)))
+
+# TS 102 226 table 5.14
+ExpandedImmediateActionResp = Struct('_tag'/Const(b'\x81'),
+ 'immediate_action_response'/Prefixed(BerTlvLen,
+ Enum(Int8ub, suspension_error=1)))
+
+# TS 102 226 table 5.16
+ExpandedScriptChainingResp = Struct('_tag'/Const(b'\x83'),
+ 'script_chaining_response'/Prefixed(BerTlvLen,
+ Enum(Int8ub, no_previous_script=1,
+ not_supported=2, unable_to_process=3)))
+
+# - starts with the "Number of executed command TLV objects" (table 5.10/5.13/5.15)
+# - followed by a sequence of R-APDU TLVs
+# - and/or one of the error # response TLVs
+ExpandedRemoteResp = Struct('_tag'/Const(b'\xab'),
+ 'body'/Prefixed(BerTlvLen, Struct(
+ 'num_executed'/ExpandedNumExecuted,
+ 'responses'/GreedyRange(Select(ExpandedR_APDU,
+ ExpandedBadFormat,
+ ExpandedImmediateActionResp,
+ ExpandedScriptChainingResp)))))
+
+
+def encode_expanded_cmd(apdus: Union[bytes, List[bytes]]) -> bytes:
+ """builds the Command Scripting template, TS 102 226 5.2.1, definite length coding
+
+ Args:
+ apdus: single C-APDU bytes or list of C-APDUs bytes. Each
+ C-APDU is wrapped into a C-APDU TLV- This function does not add
+ or modify Le.
+ Returns:
+ encoded Command Scripting template (AA...) as bytes
+ """
+ if isinstance(apdus, (bytes, bytearray)):
+ apdus = [apdus]
+ return ExpandedCmd.build({'commands': [{'c_apdu': b2h(a)} for a in apdus]})
+
+
+def decode_expanded_resp(data: bytes) -> Container:
+ """Decode a Response Scripting template, TS 102 226 5.2.2 definite length
+ coding
+
+ returned Container has:
+ number_of_commands -- "number of executed command TLV objects" table 5.11
+ commands -- list of Containers, one per R-APDU TLV, each
+ with 'response_data' and 'status_word' hexstr
+ last_response_data -- response_data of the last R-APDU or ''
+ last_status_word -- status_word of the last R-APDU or None
+ truncated -- True if any R-APDU has SW 62F1.
+ 5.2.1.1 states card sets that status when it had to truncate
+ C-APDU response data, and "this shall terminate the
+ processing of the command list".
+ so the response is short AND the remaining commands never ran.
+ bad_format -- error type of a trailing Bad format TLV if present
+ immediate_action_response -- Immediate Action Response TLV, if there was a suspension error
+ script_chaining_response -- Script Chaining Response TLV, if there was a chaining error
+
+ The 'last_response_data'/'last_status_word'/'number_of_commands' keys are compatible with
+ CompactRemoteResp so existing callers keep working."""
+ if isinstance(data, str):
+ data = h2b(data)
+ parsed = ExpandedRemoteResp.parse(data)
+ commands = []
+ bad_format = None
+ immediate_action_response = None
+ script_chaining_response = None
+ for item in parsed['body']['responses']:
+ if 'r_apdu' in item:
+ commands.append(Container(response_data=item['r_apdu']['response_data'],
+ status_word=item['r_apdu']['status_word']))
+ elif 'bad_format' in item:
+ bad_format = item['bad_format']
+ elif 'immediate_action_response' in item:
+ immediate_action_response = item['immediate_action_response']
+ elif 'script_chaining_response' in item:
+ script_chaining_response = item['script_chaining_response']
+ last = commands[-1] if commands else None
+ # TS 102 226 5.2.1.1: 62F1 means response of a C-APDU was truncated, processing terminated
+ truncated = any(c['status_word'].lower() == '62f1' for c in commands)
+ return Container(number_of_commands=parsed['body']['num_executed']['number_of_commands'],
+ commands=commands,
+ last_response_data=last['response_data'] if last else '',
+ last_status_word=last['status_word'] if last else None,
+ truncated=truncated,
+ bad_format=bad_format,
+ immediate_action_response=immediate_action_response,
+ script_chaining_response=script_chaining_response)
+
RC_CC_DS = Enum(BitsInteger(2), no_rc_cc_ds=0, rc=1, cc=2, ds=3)
CNTR_REQ = Enum(BitsInteger(2), no_counter=0, counter_no_replay_or_seq=1, counter_must_be_higher=2, counter_must_be_lower=3)
POR_REQ = Enum(BitsInteger(2), no_por=0, por_required=1, por_only_when_error=2)
@@ -149,13 +318,23 @@
raise ValueError("Invalid rc_cc_ds: %s" % spi['rc_cc_ds'])
@abc.abstractmethod
- def encode_cmd(self, otak: OtaKeyset, tar: bytes, spi: dict, apdu: bytes) -> bytes:
+ def encode_cmd(self, otak: OtaKeyset, tar: bytes, spi: dict,
+ apdu: Union[bytes, List[bytes]], remote_format: str = 'compact') -> bytes:
+ """Encode a command for a format.
+
+ remote_format:
+ 'compact' TS 102 226 5.1, DEFAULT assumes apdus are opaque already-concatenated command strings
+ 'expanded' TS 102 226 5.2 wraps a single C-APDU or list of C-APDUs in a Command Scripting template."""
pass
@abc.abstractmethod
- def decode_resp(self, otak: OtaKeyset, spi: dict, apdu: bytes) -> (object, Optional["CompactRemoteResp"]):
- """Decode a response into a response packet and, if indicted (by a
- response status of `"por_ok"`) a decoded response.
+ def decode_resp(self, otak: OtaKeyset, spi: dict, apdu: bytes,
+ remote_format: str = 'compact') -> (object, Optional[object]):
+ """Decode response into response packet + a decoded response if por_ok.
+
+ remote_format:
+ 'compact' -> DEFAULT TS 102 226 5.1.2 CompactRemoteResp2
+ 'expanded' -> container returned by decode_expanded_resp(), TS 102 226 5.2.2
The response packet's common characteristics are not fully determined,
and (so far) completely proprietary per dialect."""
@@ -335,7 +514,16 @@
'secured_data'/GreedyBytes)
hdr_construct = Struct('chl'/Int8ub, 'spi'/SPI, 'kic'/KIC, 'kid'/KID_CC, 'tar'/Bytes(3))
- def encode_cmd(self, otak: OtaKeyset, tar: bytes, spi: dict, apdu: bytes) -> bytes:
+ def encode_cmd(self, otak: OtaKeyset, tar: bytes, spi: dict,
+ apdu: Union[bytes, List[bytes]], remote_format: str = 'compact') -> bytes:
+ # as above:
+ # expanded format is a Command Scripting template wrapping the C-APDU(s)
+ # compact format passes already concatenated command string
+ if remote_format == 'expanded':
+ apdu = encode_expanded_cmd(apdu)
+ elif remote_format != 'compact':
+ raise ValueError("Invalid remote_format: %s" % remote_format)
+
# length of signature in octets
len_sig = self._compute_sig_len(spi)
pad_cnt = 0
@@ -446,7 +634,10 @@
return hdr_dec['tar'], spi, apdu
- def decode_resp(self, otak: OtaKeyset, spi: dict, data: bytes) -> ("OtaDialectSms.SmsResponsePacket", Optional["CompactRemoteResp"]):
+ def decode_resp(self, otak: OtaKeyset, spi: dict, data: bytes,
+ remote_format: str = 'compact') -> ("OtaDialectSms.SmsResponsePacket", Optional[object]):
+ if remote_format not in ('compact', 'expanded'):
+ raise ValueError("Invalid remote_format: %s ?!" % remote_format)
if isinstance(data, str):
data = h2b(data)
# plain-text POR: 027100000e0ab000110000000000000001612f
@@ -492,9 +683,11 @@
else:
raise OtaCheckError('Unknown por_rc_cc_ds: %s' % spi['por_rc_cc_ds'])
- # TODO: ExpandedRemoteResponse according to TS 102 226 5.2.2
if res.response_status == 'por_ok' and len(res['secured_data']):
- dec = CompactRemoteResp.parse(res['secured_data'])
+ if remote_format == 'expanded':
+ dec = decode_expanded_resp(res['secured_data'])
+ else:
+ dec = CompactRemoteResp.parse(res['secured_data'])
else:
dec = None
return (res, dec)
diff --git a/tests/unittests/test_ota.py b/tests/unittests/test_ota.py
index 888f828..e7c7230 100644
--- a/tests/unittests/test_ota.py
+++ b/tests/unittests/test_ota.py
@@ -300,5 +300,214 @@
self.assertEqual(d.last_status_word, t['response']['last_status_word'])
self.assertEqual(d.last_response_data, t['response']['last_response_data'])
+
+######################################################################
+# Expanded Remote Application data format (ETSI TS 102 226 Section 5.2)
+######################################################################
+
+class BerTlvLengthTestCase(unittest.TestCase):
+ """The definite-length BER-TLV length field (ISO/IEC 8825-1) used by the
+ expanded format, incl. the multi-byte (>127) forms (0x81xx / 0x82xxxx)."""
+ def test_roundtrip(self):
+ # (length value, expected encoded bytes)
+ vectors = [
+ (0, '00'),
+ (1, '01'),
+ (127, '7f'),
+ (128, '8180'),
+ (198, '81c6'), # big ~198 byte GET STATUS registry from a sja5
+ (255, '81ff'),
+ (256, '820100'),
+ (65535, '82ffff'),
+ ]
+ for length, encoded in vectors:
+ with self.subTest(length=length):
+ built = BerTlvLen.build(length)
+ self.assertEqual(b2h(built), encoded)
+ self.assertEqual(BerTlvLen.parse(built), length)
+
+
+class ExpandedCmdTestCase(unittest.TestCase):
+ """Command Scripting template TS 102 226 5.2.1"""
+
+ def test_single_capdu_golden(self):
+ # GP GET STATUS, Le=00, TS 102 226 5.2.1.1 R-APDU
+ out = encode_expanded_cmd(h2b('80f24002024f0000'))
+ # aa = TS 101 220 table 7.18 Command Scripting template tag
+ # 0a = length 10
+ # 22 = TS 101 220 table 7.19 C-APDU tag
+ # 08 = length
+ # + C-APDU
+ self.assertEqual(b2h(out), 'aa0a220880f24002024f0000')
+
+ def test_multi_capdu_golden(self):
+ out = encode_expanded_cmd([h2b('80f24002024f0000'), h2b('00a40004023f0000')])
+ self.assertEqual(b2h(out), 'aa14220880f24002024f0000220800a40004023f0000')
+
+ def test_multibyte_length_golden(self):
+ # C-APDU: 4 header + 1 Lc + 195 data = 200 bytes.
+ # 200 byte C-APDU forces long form BER lengths:
+ # C-APDU TLV, 200 -> 81c8 + template 203 -> 81cb
+ capdu = h2b('80f24000') + bytes([195]) + bytes(range(195))
+ self.assertEqual(len(capdu), 200)
+ out = encode_expanded_cmd(capdu)
+ # aa 81 cb | 22 81 c8 | <200 byte capdu>
+ self.assertEqual(b2h(out[:6]), 'aa81cb2281c8')
+ self.assertEqual(out[6:], capdu)
+
+ def test_roundtrip(self):
+ for apdus in [[h2b('80f24002024f0000')],
+ [h2b('00a40004023f00'), h2b('80f24002024f0000')],
+ [h2b('00'*250)]]:
+ with self.subTest(n=len(apdus)):
+ out = encode_expanded_cmd(apdus)
+ parsed = ExpandedCmd.parse(out)
+ self.assertEqual([h2b(c.c_apdu) for c in parsed.commands], apdus)
+
+
+class ExpandedRespTestCase(unittest.TestCase):
+ """Decoding of the Response Scripting template (TS 102 226 5.2.2)."""
+
+ def test_registry_golden(self):
+ # real card case: GET STATUS returns a ~198 byte registry TLV + SW 9000
+ # R-APDU = 198 data + 2 SW = 200/81c8
+ # 'number of executed' TLV 80 01 01.
+ registry = bytes(range(198))
+ data = ExpandedRemoteResp.build(dict(body=dict(
+ num_executed=dict(number_of_commands=1),
+ responses=[dict(r_apdu=dict(response_data=b2h(registry), status_word='9000'))])))
+ # ab | 81 ce | 80 01 01 | 23 81 c8 | <198 data> 90 00
+ self.assertEqual(b2h(data[:9]), 'ab81ce8001012381c8')
+ dec = decode_expanded_resp(data)
+ self.assertEqual(dec.number_of_commands, 1)
+ self.assertEqual(len(dec.commands), 1)
+ self.assertEqual(dec.last_status_word, '9000')
+ self.assertEqual(dec.last_response_data, b2h(registry))
+
+ def test_status_only_golden(self):
+ # last command, no response data, SW 6132
+ data = ExpandedRemoteResp.build(dict(body=dict(
+ num_executed=dict(number_of_commands=1),
+ responses=[dict(r_apdu=dict(response_data='', status_word='6132'))])))
+ self.assertEqual(b2h(data), 'ab0780010123026132')
+ dec = decode_expanded_resp(data)
+ self.assertEqual(dec.last_status_word, '6132')
+ self.assertEqual(dec.last_response_data, '')
+
+ def test_multi_command(self):
+ data = ExpandedRemoteResp.build(dict(body=dict(
+ num_executed=dict(number_of_commands=2),
+ responses=[dict(r_apdu=dict(response_data='6f21', status_word='9000')),
+ dict(r_apdu=dict(response_data='', status_word='6a82'))])))
+ dec = decode_expanded_resp(data)
+ self.assertEqual(dec.number_of_commands, 2)
+ self.assertEqual([(c.status_word, c.response_data) for c in dec.commands],
+ [('9000', '6f21'), ('6a82', '')])
+ # last == final R-APDU, error status included
+ self.assertEqual(dec.last_status_word, '6a82')
+ self.assertEqual(dec.last_response_data, '')
+
+ def test_bad_format(self):
+ # ab | 06 | 80 01 01 | 90 01 01
+ data = h2b('ab06800101900101')
+ dec = decode_expanded_resp(data)
+ self.assertEqual(str(dec.bad_format), 'unknown_tag')
+ self.assertIsNone(dec.last_status_word)
+
+ def test_immediate_action_error(self):
+ # ab | 06 | 80 01 01 | 81 01 01
+ data = h2b('ab06800101810101')
+ dec = decode_expanded_resp(data)
+ self.assertEqual(str(dec.immediate_action_response), 'suspension_error')
+
+ def test_script_chaining_error(self):
+ # ab | 06 | 80 01 01 | 83 01 02
+ data = h2b('ab06800101830102')
+ dec = decode_expanded_resp(data)
+ self.assertEqual(str(dec.script_chaining_response), 'not_supported')
+
+ def test_truncation_is_flagged(self):
+ """TS 102 226 5.2.1.1: SW 62F1 means the C-APDU response data was truncated, and
+ "this shall terminate the processing of the command list"
+ halves are invisible in the R-APDU list, truncated + aborted script must not pass as complete"""
+ # second command truncated -> processing stopped at that point
+ data = ExpandedRemoteResp.build(dict(body=dict(
+ num_executed=dict(number_of_commands=2),
+ responses=[dict(r_apdu=dict(response_data='6f21', status_word='9000')),
+ dict(r_apdu=dict(response_data='aabb', status_word='62f1'))])))
+ dec = decode_expanded_resp(data)
+ self.assertTrue(dec.truncated)
+ self.assertEqual(dec.last_status_word, '62f1')
+
+ def test_untruncated_response_is_not_flagged(self):
+ data = ExpandedRemoteResp.build(dict(body=dict(
+ num_executed=dict(number_of_commands=1),
+ responses=[dict(r_apdu=dict(response_data='6f21', status_word='9000'))])))
+ self.assertFalse(decode_expanded_resp(data).truncated)
+ # 62xx that is not 62F1 is warning, not truncation
+ data = ExpandedRemoteResp.build(dict(body=dict(
+ num_executed=dict(number_of_commands=1),
+ responses=[dict(r_apdu=dict(response_data='', status_word='6282'))])))
+ self.assertFalse(decode_expanded_resp(data).truncated)
+
+
+class ExpandedSmsPipelineTestCase(unittest.TestCase):
+ """expanded format + TS 102 225 SMS security witj 3DES keyset,
+ to ensure remote_format does not affect the compact path"""
+ def __init__(self, methodName='runTest', **kwargs):
+ super().__init__(methodName, **kwargs)
+ self.od = OtaKeyset(algo_crypt='triple_des_cbc2', kic_idx=3,
+ kic=h2b('C21DD66ACAC13CB3BC8B331B24AFB57B'),
+ algo_auth='triple_des_cbc2', kid_idx=3,
+ kid=h2b('12110C78E678C25408233076AA033615'))
+ self.dialect = OtaDialectSms()
+ self.tar = h2b('000000')
+
+ def test_cmd_expanded_secured_roundtrip(self):
+ spi = SPI_CC_POR_CIPHERED_CC
+ enc = self.dialect.encode_cmd(self.od, self.tar, spi, h2b('80f24002024f0000'),
+ remote_format='expanded')
+ # decode_cmd returns opaque 'Command Scripting template'
+ dec_tar, dec_spi, dec_secured = self.dialect.decode_cmd(self.od, enc)
+ self.assertEqual(b2h(dec_tar), b2h(self.tar))
+ self.assertEqual(dec_spi, spi)
+ self.assertEqual(b2h(dec_secured), 'aa0a220880f24002024f0000')
+
+ def test_cmd_expanded_list(self):
+ spi = SPI_CC_POR_CIPHERED_CC
+ enc = self.dialect.encode_cmd(self.od, self.tar, spi,
+ [h2b('80f24002024f0000'), h2b('00a40004023f0000')],
+ remote_format='expanded')
+ _, _, dec_secured = self.dialect.decode_cmd(self.od, enc)
+ parsed = ExpandedCmd.parse(dec_secured)
+ self.assertEqual([c.c_apdu for c in parsed.commands],
+ ['80f24002024f0000', '00a40004023f0000'])
+
+ def test_resp_expanded_plaintext(self):
+ # plaintext (u:nciphered + no CC) expanded response SMS
+ # containing a 198 byte GP registry + SW 9000 as above, decode it through decode_resp().
+ spi = SPI_CC_POR_UNCIPHERED_NOCC
+ registry = bytes(range(198))
+ secured = ExpandedRemoteResp.build(dict(body=dict(
+ num_executed=dict(number_of_commands=1),
+ responses=[dict(r_apdu=dict(response_data=b2h(registry), status_word='9000'))])))
+ rpl = 1 + 3 + 5 + 1 + 1 + len(secured) # RHL-STS + secured data
+ resp_body = rpl.to_bytes(2, 'big') + b'\x0a' + self.tar + b'\x00'*5 + b'\x00' + b'\x00' + secured
+ sms = b'\x02\x71\x00' + resp_body
+ r, dec = self.dialect.decode_resp(self.od, spi, sms, remote_format='expanded')
+ self.assertEqual(r.response_status, 'por_ok')
+ self.assertEqual(dec.number_of_commands, 1)
+ self.assertEqual(dec.last_status_word, '9000')
+ self.assertEqual(dec.last_response_data, b2h(registry))
+
+ def test_compact_still_default(self):
+ # no remote_format -> compact default
+ spi = SPI_CC_POR_UNCIPHERED_NOCC
+ r, d = self.dialect.decode_resp(self.od, spi, '027100000e0ab000110000000000000001612f')
+ self.assertEqual(d.number_of_commands, 1)
+ self.assertEqual(d.last_status_word, '612f')
+ self.assertEqual(d.last_response_data, '')
+
+
if __name__ == "__main__":
unittest.main()
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/43541?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: Idec00d16fd1a7d4a7129b2a3b6f0ef37dabcecb7
Gerrit-Change-Number: 43541
Gerrit-PatchSet: 1
Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>
1
0
[M] Change in pysim[master]: smpp-ota-tool: add --format compact,expanded for TS 102 226 5.2
by Hoernchen 08 Sep '26
by Hoernchen 08 Sep '26
08 Sep '26
Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/43542?usp=email )
Change subject: smpp-ota-tool: add --format compact,expanded for TS 102 226 5.2
......................................................................
smpp-ota-tool: add --format compact,expanded for TS 102 226 5.2
Add --format expanded flag to send C-APDUs in the expanded remote
application data format.
Each --apdu becomes its own C-APDU TLV, and the tool
logs the full per-command R-APDU list decoded from the Response Scripting
template and warns if the card reports a truncated response.
Default stays 'compact', unchanged.
Change-Id: Idad90756f85bb7e54ef720f6067bb5d6dcff0c42
---
M contrib/smpp-ota-tool.py
M docs/smpp-ota-tool.rst
2 files changed, 64 insertions(+), 7 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/42/43542/1
diff --git a/contrib/smpp-ota-tool.py b/contrib/smpp-ota-tool.py
index 28c13bf..3698001 100755
--- a/contrib/smpp-ota-tool.py
+++ b/contrib/smpp-ota-tool.py
@@ -70,6 +70,8 @@
option_parser.add_argument('--src-addr', default='12', type=str, help='SMS source address (MSISDN)')
option_parser.add_argument('--dest-addr', default='23', type=str, help='SMS destination address (MSISDN)')
option_parser.add_argument('--timeout', default=10, type=int, help='Maximum response waiting time')
+option_parser.add_argument('--format', choices=['compact', 'expanded'], default='compact',
+ help="Remote Application data format: 'compact' or 'expanded'")
option_parser.add_argument('-a', '--apdu', action='append', required=True, type=is_hexstr, help='C-APDU to send')
class SmppHandler:
@@ -77,7 +79,8 @@
def __init__(self, host: str, port: int,
system_id: str, password: str,
- ota_keyset: OtaKeyset, spi: dict, tar: bytes):
+ ota_keyset: OtaKeyset, spi: dict, tar: bytes,
+ remote_format: str = 'compact'):
"""
Initialize connection to SMPP server and set static OTA SMS-TPDU ciphering parameters
Args:
@@ -88,6 +91,7 @@
ota_keyset: OTA keyset to be used for SMS-TPDU ciphering
spi: Security Parameter Indicator (SPI) to be used for SMS-TPDU ciphering
tar: Toolkit Application Reference (TAR) of the targeted card application
+ remote_format: Remote Application data format ('compact' or 'expanded', TS 102 226)
"""
# Create and connect SMPP client
@@ -103,6 +107,7 @@
self.ota_keyset = ota_keyset
self.tar = tar
self.spi = spi
+ self.remote_format = remote_format
def __del__(self):
if self.client:
@@ -113,14 +118,16 @@
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)
+ 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)
+ 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
return None
@@ -183,10 +190,14 @@
tuple containing the last response data and the last status word as byte strings
"""
- logger.info("C-APDU sending: %s...", b2h(apdu))
+ if isinstance(apdu, (list, tuple)):
+ logger.info("C-APDU(s) sending: %s...", [b2h(a) for a in apdu])
+ else:
+ logger.info("C-APDU sending: %s...", b2h(apdu))
# translate to Secured OTA RFM
- secured = self.ota_dialect.encode_cmd(self.ota_keyset, self.tar, self.spi, apdu=apdu)
+ secured = self.ota_dialect.encode_cmd(self.ota_keyset, self.tar, self.spi, apdu=apdu,
+ remote_format=self.remote_format)
# add user data header
tpdu = b'\x02\x70\x00' + secured
# send via SMPP
@@ -200,6 +211,17 @@
container_dict = dict(container)
resp = container_dict.get('last_response_data')
sw = container_dict.get('last_status_word')
+ # expanded format: decoded response carries
+ # per command R-APDU list; log each one.
+ for i, cmd in enumerate(container_dict.get('commands') or []):
+ logger.info("R-APDU[%u] received: %s %s", i,
+ cmd['response_data'], cmd['status_word'])
+ if container_dict.get('truncated'):
+ logger.warning("Response was TRUNCATED (SW 62F1): the card cut the response "
+ "data short and did not execute the rest of the script")
+ if container_dict.get('bad_format') is not None:
+ logger.warning("Response contains a Bad format TLV: %s",
+ container_dict['bad_format'])
if resp is None:
raise ValueError("Response does not contain any last_response_data, no R-APDU received!")
if sw is None:
@@ -233,8 +255,14 @@
'por_shall_be_ciphered': not opts.por_no_ciphering,
'por_rc_cc_ds': opts.por_rc_cc_ds,
'por': opts.por_req}
- apdu = h2b("".join(opts.apdu))
+ if opts.format == 'expanded':
+ # TS 102 226 5.2.1.1: wrap each apdu in its own C-APDU TLV
+ apdu = [h2b(a) for a in opts.apdu]
+ else:
+ # compact: C-APDUs are concatenated as single command string
+ apdu = h2b("".join(opts.apdu))
- smpp_handler = SmppHandler(opts.host, opts.port, opts.system_id, opts.password, ota_keyset, spi, h2b(opts.tar))
+ smpp_handler = SmppHandler(opts.host, opts.port, opts.system_id, opts.password, ota_keyset, spi,
+ h2b(opts.tar), remote_format=opts.format)
resp, sw = smpp_handler.transceive_apdu(apdu, opts.src_addr, opts.dest_addr, opts.timeout)
print("%s %s" % (b2h(resp), b2h(sw)))
diff --git a/docs/smpp-ota-tool.rst b/docs/smpp-ota-tool.rst
index beb494a..9dc0df6 100644
--- a/docs/smpp-ota-tool.rst
+++ b/docs/smpp-ota-tool.rst
@@ -170,6 +170,35 @@
.. note:: The replay-protection-counter is implemented as a 5 byte integer value (see also ETSI TS 102 225, Table 3).
When the counter has reached its maximum, it will not overflow nor can it be reset.
+Expanded remote application data format
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+`smpp-ota-tool` uses the TS 102 226 section 5.1 compact remote application data format by default. This
+format concatenates C-APDUs into one command string and only the result of the LAST executed command is reported back.
+Retrieving the response data therefore requires a GET RESPONSE C-APDU, and only a single GET RESPONSE command may occur per script.
+
+The TS 102 226 section 5.2 expanded remote application data format removes these limitations: Each C-APDU is
+wrapped in its own C-APDU TLV inside a Command Scripting template, and the response is a Response Scripting template that contains one R-APDU TLV with the full response data and status word per executed command. To use it, pass
+``--format expanded``; every ``--apdu`` argument then becomes its own C-APDU TLV.
+
+.. note:: The expanded format does not use GET RESPONSE. To retrieve response data from a case 2 or case 4
+ command, include an ``Le`` field in the C-APDU. i.e. ``Le='00'`` instructs the card to return all available
+ response data in the R-APDU, with no 256-byte limit (TS 102 226, section 5.2.1.1). Without the ``Le``
+ field no response data is returned, except a status word for the last command!.
+
+For example, a GP GET STATUS of all applications (``80F24002024F00``) returns a registry that can be much
+larger than 256 bytes. In the compact format the card would only answer with ``61xx`` procedure bytes. In the expanded
+format, appending ``Le='00'`` (i.e. ``80F24002024F0000``) makes the card return the whole registry in one exchange:
+
+::
+
+ $ PYTHONPATH=./ ./contrib/smpp-ota-tool.py --kic <KIC> --kid <KID> --kid-idx 1 --kic-idx 1 \
+ --algo-crypt triple_des_cbc2 --algo-auth triple_des_cbc2 --tar 000000 --cntr-req no_counter \
+ --format expanded --apdu 80F24002024F0000
+
+The response data (a concatenation of GlobalPlatform registry TLVs) can then be decoded with
+``pySim.global_platform.GpRegistryRelatedData.from_tlv()``.
+
smpp-ota-tool syntax
~~~~~~~~~~~~~~~~~~~~
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/43542?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: Idad90756f85bb7e54ef720f6067bb5d6dcff0c42
Gerrit-Change-Number: 43542
Gerrit-PatchSet: 1
Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>
1
0
[L] Change in pysim[master]: sms/smpp-ota-tool: reassemble multi part response SMS
by Hoernchen 08 Sep '26
by Hoernchen 08 Sep '26
08 Sep '26
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(a)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')
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/43544?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: I8c81097e607e0d055c4f031bbcc8a74d5c24a0e7
Gerrit-Change-Number: 43544
Gerrit-PatchSet: 1
Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>
1
0
[M] Change in pysim[master]: transport/smpp2sim: TERMINAL RESPONSE for proactive SEND SHORT MESSAGE
by Hoernchen 08 Sep '26
by Hoernchen 08 Sep '26
08 Sep '26
Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/43543?usp=email )
Change subject: transport/smpp2sim: TERMINAL RESPONSE for proactive SEND SHORT MESSAGE
......................................................................
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
---
M pySim/bip.py
M pySim/transport/__init__.py
A tests/unittests/test_transport.py
3 files changed, 79 insertions(+), 2 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/43/43543/1
diff --git a/pySim/bip.py b/pySim/bip.py
index 287f31d..5fc83ce 100644
--- a/pySim/bip.py
+++ b/pySim/bip.py
@@ -102,6 +102,14 @@
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."""
diff --git a/pySim/transport/__init__.py b/pySim/transport/__init__.py
index d689a0a..4fb38f1 100644
--- a/pySim/transport/__init__.py
+++ b/pySim/transport/__init__.py
@@ -70,10 +70,24 @@
raise NotImplementedError('No handler method for %s' % pcmd.decoded)
def prepare_response(self, pcmd: ProactiveCommand, general_result: str = 'performed_successfully'):
+ # 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'],
diff --git a/tests/unittests/test_transport.py b/tests/unittests/test_transport.py
new file mode 100644
index 0000000..b2c7f08
--- /dev/null
+++ b/tests/unittests/test_transport.py
@@ -0,0 +1,55 @@
+#!/usr/bin/env python3
+
+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):
+ # Checkk 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[1].to_tlv()), '82028381')
+
+ 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()
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/43543?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: Ib96ce81c4ff093b8a6fc715f79617e95a2dd8433
Gerrit-Change-Number: 43543
Gerrit-PatchSet: 1
Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>
1
0
08 Sep '26
Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/43545?usp=email )
Change subject: smpp2sim: make the SCP81 BIP relay work
......................................................................
smpp2sim: make the SCP81 BIP relay work
The BIP relay ( the "handset" side for SCP81) never worked: the
connect callback in handle_OpenChannel was "never called" as the fixme
says, everything else was missing.
Fixme cause: card APDU I/O is driven synchronously, proactive command loop
lives in a blocking while loop (pySim.transport.LinkBase.send_apdu_checksw)
that runs on the Twisted reactor thread. A Twisted TCP4ClientEndpoint +
connectProtocol only completes when the reactor does reactor things,
but the reactor thread is stuck in that loop for the whole proactive
session...
Fixme fix: don't fight the reactor, just drive the relay channel with a plain
old blocking socket, which fits the synchronous execution model.
Channel numbers now come from the command Device identities (channel_N ->
low nibble) instead of the hard coded chan_nr == 1.
Additionally fix two bugs found on the path to scp81 glory:
- TERMINAL RESPONSE device identities are forced to terminal->UICC per
TS 102 223 6.8.2 (prepare_response() inverts the command identities,
which for a channel-addressed BIP command yields channel_N->UICC).
- Error responses now build a valid AddlInfoBip cause, prepare_response()
hard coded empty "additional information" cannot be encoded for a
BIP error.
And some tests based on real card interactions.
Change-Id: If96c768f2e35c20ea3753e601059410121517b60
---
M pySim-smpp2sim.py
M pySim/bip.py
M tests/unittests/test_bip_relay.py
3 files changed, 533 insertions(+), 40 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/45/43545/1
diff --git a/pySim-smpp2sim.py b/pySim-smpp2sim.py
index 2c7ff51..f7a45f5 100755
--- a/pySim-smpp2sim.py
+++ b/pySim-smpp2sim.py
@@ -30,10 +30,13 @@
import argparse
import logging
+import socket
+import threading
+import time
import colorlog
from twisted.protocols import basic
-from twisted.internet import defer, endpoints, protocol, reactor, task
+from twisted.internet import defer, endpoints, reactor, task
from twisted.cred.portal import IRealm
from twisted.cred.checkers import InMemoryUsernamePasswordDatabaseDontUse
from twisted.cred.portal import Portal
@@ -55,6 +58,7 @@
from pySim.cat import ProactiveCommand, SendShortMessage, SMS_TPDU, SMSPPDownload, BearerDescription
from pySim.cat import DeviceIdentities, Address, OtherAddress, UiccTransportLevel, BufferSize
from pySim.cat import ChannelStatus, ChannelData, ChannelDataLength
+from pySim.cat import EventList, EventDownload, Result
from pySim.utils import b2h, h2b
logger = logging.getLogger(__name__)
@@ -72,18 +76,6 @@
print("-> %s %s" % (cmd[:10], cmd[10:]))
print("<- %s: %s" % (sw, resp))
-class TcpProtocol(protocol.Protocol):
- def dataReceived(self, data):
- pass
-
- def connectionLost(self, reason):
- pass
-
-
-def tcp_connected_callback(p: protocol.Protocol):
- """called by twisted TCP client."""
- logger.error("%s: connected!" % p)
-
def dcs_is_8bit(dcs):
if dcs == pdu_types.DataCoding(pdu_types.DataCodingScheme.DEFAULT,
pdu_types.DataCodingDefault.OCTET_UNSPECIFIED):
@@ -118,6 +110,11 @@
smppEndpoint = endpoints.TCP6ServerEndpoint(reactor, tcp_port, interface=bind_ip)
smppEndpoint.listen(self.factory)
self.tp = self.scc = self.card = None
+ # Serialise card/APDU access.
+ # - SMPP handler drives the card from reactor thread
+ # - BIP relay data-available path drives it from socket reader thread.
+ # The transport is not re-entrant, both must take this lock.
+ self._card_lock = threading.Lock()
def connect_to_card(self, tp: LinkBase):
self.tp = tp
@@ -130,6 +127,21 @@
self.card.select_adf_by_aid(adf='usim')
# FIXME: create a more realistic profile than ffffff
self.scc.terminal_profile('ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff')
+ # Connect the BIP relay inbound path to the card.
+ # relay socket receives data -> ME initiated ENVELOPE EVENT DOWNLOA
+ # -> triggers RECEIVE DATA proactive session.
+ # FIXME this cross-thread push to the card is exercised only with real hardware
+ # the card free tests cover socket relay + envelope construction, not delivery.
+ handler = getattr(tp, 'proactive_handler', None)
+ if isinstance(handler, Proact):
+ handler.data_available_sink = self._deliver_data_available
+
+ def _deliver_data_available(self, envelope_hex: str):
+ """push ME initiated ENVELOPE EVENT DOWNLOAD to the card"""
+ with self._card_lock:
+ logger.info("ENVELOPE(Data available): %s" % envelope_hex)
+ (data, sw) = self.scc.envelope(envelope_hex)
+ logger.info("SW %s: %s" % (sw, data))
def _msgHandler(self, system_id, smpp, pdu):
"""Handler for incoming messages received via SMPP from ESME."""
@@ -164,7 +176,8 @@
# 3) send to the card
envelope_hex = b2h(sms_dl.to_tlv())
logger.info("ENVELOPE: %s" % envelope_hex)
- (data, sw) = self.scc.envelope(envelope_hex)
+ with self._card_lock:
+ (data, sw) = self.scc.envelope(envelope_hex)
logger.info("SW %s: %s" % (sw, data))
if sw in ['9200', '9300']:
# TODO send back RP-ERROR message with TP-FCS == 'SIM Application Toolkit Busy'
diff --git a/pySim/bip.py b/pySim/bip.py
index 5fc83ce..e5ec029 100644
--- a/pySim/bip.py
+++ b/pySim/bip.py
@@ -17,6 +17,19 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
+# A ProactiveHandler with TCP sockets that backs the BIP channels,
+# so a card can run its own IP session (SCP81/HTTPS, CAT_TP, ...)
+#
+# Currently used by pySim-smpp2sim.py which connects the SMS path to its SMPP server.
+# Other drivers can pass their own sinks:
+#
+# handler = Proact(data_available_sink=..., sms_sink=...)
+# tp = init_reader(opts, proactive_handler=handler)
+#
+# Both sinks are optional.
+# Without them the handler builds and logs what it would send.
+
+
import logging
import socket
import threading
@@ -34,43 +47,161 @@
logger = logging.getLogger(__name__)
class ProactChannel:
- """Representation of a single protective channel."""
+ """Representation of a single BIP channel, backed by a blocking TCP
+ socket.
+
+ Why blocking sockets and not Twisted endpoints, considering we have
+ twisted?
+ The proactive-command loop lives in a blocking while-loop,
+ "pySim.transport.LinkBase.send_apdu_checksw" that runs on the
+ Twisted reactor thread. A Twisted async TCP client only makes any
+ progress when the reactor uhh... reacts, but the reactor is stuck
+ in that loop for the whole proactive session -> the connectProtocol()
+ Deferred never fires while we are handling OPEN/SEND/RECEIVE CHANNEL.
+ Plain blocking sockets just work: connect() in handle_OpenChannel,
+ send() in handle_SendData, recv() feeding a buffer for handle_ReceiveData.
+ No need to make it harder than it has to be to handle the "massive" T0
+ bandwidth..
+
+ While TLS runs on the card and is the thing we are actually interested
+ in this only moves opaque bytes between the card and the socket, and
+ never looks at the payload.
+ """
+ # how much we try to read off the socket per recv()
+ RECV_CHUNK = 4096
+
def __init__(self, channels: 'ProactChannels', chan_nr: int):
self.channels = channels
self.chan_nr = chan_nr
- self.ep = None
+ self.sock = None
+ # TS 102 223 says the terminal keeps an Rx buffer per channel; RECEIVE
+ # DATA drains it, and it is filled asynchronously as the peer sends.
+ self.rx_buf = bytearray()
+ self._rx_lock = threading.Lock()
+ self._reader = None
+ self._closing = False
+ self.peer_closed = False
+
+ def connect(self, host: str, port: int, timeout: float = 10.0):
+ """Open the blocking TCP socket and start the background Rx reader."""
+ s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+ s.settimeout(timeout)
+ s.connect((host, port))
+ # Back to blocking mode for the reader thread.
+ # CLOSE CHANNEL unblocks the pending recv() via shutdown().
+ s.settimeout(None)
+ self.sock = s
+ self._reader = threading.Thread(target=self._rx_loop,
+ name='bip-rx-%d' % self.chan_nr, daemon=True)
+ self._reader.start()
+
+ def _rx_loop(self):
+ """Continuously read from the socket into rx_buf, like a real ME.
+ TS 102 223 7.5.10.1 says the event is raised 'only if the targeted channel
+ buffer is empty when new data arrives in it'
+ So we fire the channels data available hook on the empty->non-empty transition """
+ while not self._closing:
+ try:
+ data = self.sock.recv(self.RECV_CHUNK)
+ except (OSError, ValueError):
+ break
+ if not data:
+ self.peer_closed = True
+ break
+ with self._rx_lock:
+ was_empty = len(self.rx_buf) == 0
+ self.rx_buf.extend(data)
+ if was_empty and not self._closing:
+ self.channels.notify_data_available(self)
+
+ def send(self, data: bytes):
+ """Tx, write bytes to the socket == SEND DATA"""
+ self.sock.sendall(data)
+
+ def rx_available(self) -> int:
+ with self._rx_lock:
+ return len(self.rx_buf)
+
+ def take_rx(self, n: int):
+ """Remove+return up to n bytes from rxbuf + remaining bytes"""
+ with self._rx_lock:
+ chunk = bytes(self.rx_buf[:n])
+ del self.rx_buf[:n]
+ remaining = len(self.rx_buf)
+ return chunk, remaining
+
+ def wait_rx(self, timeout: float) -> int:
+ """wait up to timeout seconds until the rxbuf has data
+ returns the number of bytes available
+ Cards have a "data available" event, card free callers use
+ this to wait for the echoed bytes."""
+ deadline = time.monotonic() + timeout
+ while time.monotonic() < deadline:
+ avail = self.rx_available()
+ if avail or self.peer_closed:
+ return avail
+ time.sleep(0.005)
+ return self.rx_available()
def close(self):
- """Close the channel."""
- if self.ep:
- self.ep.disconnect()
+ """Close channel: stop reader, close socket, drop bookkeeping."""
+ self._closing = True
+ if self.sock is not None:
+ try:
+ self.sock.shutdown(socket.SHUT_RDWR)
+ except OSError:
+ pass
+ try:
+ self.sock.close()
+ except OSError:
+ pass
+ # CLOSE CHANNEL synchronously handled inside the rx reader thread
+ # (data-available -> ENVELOPE -> FETCH -> handle_CloseChannel -> close),
+ # so close() can be called on the reader thread.
+ # Joining self raises "cannot join current thread" so better skip i..
+ # setting _closing + shutting down the socket already makes _rx_loop
+ # return on the next iteration anyway.
+ if self._reader is not None and self._reader is not threading.current_thread():
+ self._reader.join(timeout=1.0)
self.channels.channel_delete(self.chan_nr)
class ProactChannels:
"""Wrapper class for maintaining state of proactive channels."""
- def __init__(self):
+ def __init__(self, on_data_available=None):
self.channels = {}
+ # called from a channel rx reader thread on empty->non-empty buf
+ # transition, with the ProactChannel as its parameter
+ self._on_data_available = on_data_available
def channel_create(self) -> ProactChannel:
"""Create a new proactive channel, allocating its integer number."""
- for i in range(1, 9):
+ for i in range(1, 8):
if not i in self.channels:
self.channels[i] = ProactChannel(self, i)
return self.channels[i]
raise ValueError('Cannot allocate another channel: All channels active')
def channel_delete(self, chan_nr: int):
- del self.channels[chan_nr]
+ self.channels.pop(chan_nr, None)
+
+ def notify_data_available(self, chan: ProactChannel):
+ if self._on_data_available:
+ self._on_data_available(chan)
class Proact(ProactiveHandler):
#def __init__(self, smpp_factory):
# self.smpp_factory = smpp_factory
- def __init__(self, sms_sink=None):
+ def __init__(self, data_available_sink=None, sms_sink=None):
+ # data_available_sink(envelope_hex) delivers ENVELOPE EVENT DOWNLOAD
+ # to the card when socket data arrives
+ # None in card free/test mode
+ self.data_available_sink = data_available_sink
# sms_sink(pdu) delivers an MO-SMS the card sent (SEND SHORT MESSAGE)
# onwards; pySim-smpp2sim.py hands it to its SMPP server. None -> log
# and drop.
self.sms_sink = sms_sink
- self.channels = ProactChannels()
+ self.channels = ProactChannels(on_data_available=self._on_channel_data_available)
@staticmethod
def _find_first_element_of_type(instlist, cls):
@@ -79,6 +210,80 @@
return i
return None
+ @staticmethod
+ def _channel_nr_from_dev_ids(dev_id_ie: DeviceIdentities) -> int:
+ """Maps id like channel_1 -> channel number.
+ TS 102 223 Section 8.7 says low nibble is channel number,
+ channel-N = 0x21..0x27"""
+ dest = dev_id_ie.decoded['dest_dev_id']
+ return DeviceIdentities.DEV_IDS.inverse[dest] & 0x0f
+
+ def _channel_for(self, dev_id_ie: DeviceIdentities):
+ """Resolve the ProactChannel addressed by a command dev id, or None"""
+ return self.channels.channels.get(self._channel_nr_from_dev_ids(dev_id_ie), None)
+
+ @staticmethod
+ def _channel_status(chan_nr: int, established: bool = True) -> str:
+ """TS 102 223 Section 8.56 channel status value for the
+ default/network bearer:
+ - byte 3 low 3 bits = channel id
+ - bit 8 = link established
+ - byte 4 = 00 no further info"""
+ b3 = (0x80 if established else 0x00) | (chan_nr & 0x07)
+ return '%02x00' % b3
+
+ def _bip_response_head(self, pcmd: ProactiveCommand,
+ general_result: str = 'performed_successfully',
+ additional_information: str = ''):
+ """CommandDetails / DeviceIdentities / Result head part of a BIP TERMINAL
+ RESPONSE. Built on prepare_response() but with two changes:
+
+ - Device identities forced source=terminal, dest=UICC.
+ TS 102 223 6.8.2 mandates for every TERMINAL RESPONSE
+ prepare_response() inverts the commands device id, which is
+ right for a uicc->terminal command but would yield a wrong
+ channel_N->UICC for the channel addressed BIP commands.
+
+ - Result is recreated for non success cases. prepare_response()
+ hard codes empty "additional information", but for enum results
+ like BIP error -> AddlInfoBip the empty value cannot be encoded at
+ all, so we always ask prepare_response() for a success Result
+ and swap for a properly encoded one here."""
+ head = self.prepare_response(pcmd, 'performed_successfully')
+ for i, ie in enumerate(head):
+ if isinstance(ie, DeviceIdentities):
+ head[i] = DeviceIdentities(decoded={'source_dev_id': 'terminal',
+ 'dest_dev_id': 'uicc'})
+ elif isinstance(ie, Result) and general_result != 'performed_successfully':
+ res = Result()
+ res.from_dict({'result': {'general_result': general_result,
+ 'additional_information': additional_information}})
+ head[i] = res
+ return head
+
+ def build_data_available_envelope(self, chan: ProactChannel) -> bytes:
+ """TS 102 223 7.5.10.2 ENVELOPE EVENT DOWNLOAD
+ Event list, Device id terminal->UICC, Channel status,
+ Channel data length (bytes available or FF for > 255)."""
+ avail = min(chan.rx_available(), 0xff)
+ ed = EventDownload(children=[
+ EventList(decoded=['data_available']),
+ DeviceIdentities(decoded={'source_dev_id': 'terminal', 'dest_dev_id': 'uicc'}),
+ ChannelStatus(decoded=self._channel_status(chan.chan_nr)),
+ ChannelDataLength(decoded=avail),
+ ])
+ return ed.to_tlv()
+
+ def _on_channel_data_available(self, chan: ProactChannel):
+ """rx reader thread hook: socket data arrived while the channel buffer
+ was empty. card uses ENVELOPE EVENT DOWNLOAD + responds by FETCHing RECEIVE DATA
+ proactive command. Card free only builds and logs"""
+ envelope_hex = b2h(self.build_data_available_envelope(chan))
+ logger.info("channel %u: %u byte(s) available -> ENVELOPE(Data available) %s",
+ chan.chan_nr, chan.rx_available(), envelope_hex)
+ if self.data_available_sink:
+ self.data_available_sink(envelope_hex)
+
"""Call-back which the pySim transport core calls whenever it receives a
proactive command from the SIM."""
def handle_SendShortMessage(self, pcmd: ProactiveCommand):
@@ -136,16 +341,22 @@
raise ValueError('Unsupported protocol_type')
if other_addr_ie.decoded.get('type_of_address', None) != 'ipv4':
raise ValueError('Unsupported type_of_address')
- ipv4_bytes = h2b(other_addr_ie.decoded['address'])
- ipv4_str = '%u.%u.%u.%u' % (ipv4_bytes[0], ipv4_bytes[1], ipv4_bytes[2], ipv4_bytes[3])
+ addr_bytes = h2b(other_addr_ie.decoded['address']) if isinstance(
+ other_addr_ie.decoded['address'], str) else other_addr_ie.decoded['address']
+ ipv4_str = '%u.%u.%u.%u' % (addr_bytes[0], addr_bytes[1], addr_bytes[2], addr_bytes[3])
port_nr = transp_lvl_ie.decoded['port_number']
- print("%s:%u" % (ipv4_str, port_nr))
+ logger.info("OpenChannel: connecting to %s:%u", ipv4_str, port_nr)
channel = self.channels.channel_create()
- channel.ep = endpoints.TCP4ClientEndpoint(reactor, ipv4_str, port_nr)
- channel.prot = TcpProtocol()
- d = endpoints.connectProtocol(channel.ep, channel.prot)
- # FIXME: why is this never called despite the client showing the inbound connection?
- d.addCallback(tcp_connected_callback)
+ # yes, blocking connect()
+ try:
+ channel.connect(ipv4_str, port_nr)
+ except OSError as e:
+ logger.warning("OpenChannel: connect to %s:%u failed: %s", ipv4_str, port_nr, e)
+ self.channels.channel_delete(channel.chan_nr)
+ return self._bip_response_head(pcmd, 'bearer_independent_protocol_error',
+ 'channel_closed') + [
+ ChannelStatus(decoded=self._channel_status(channel.chan_nr, established=False)),
+ bearer_desc_ie, buffer_size_ie]
# Terminal Response example: [
# {'command_details': {'command_number': 1,
@@ -157,12 +368,22 @@
# {'bearer_description': {'bearer_type': 'default', 'bearer_parameters': ''}},
# {'buffer_size': 1024}
# ]
- return self.prepare_response(pcmd) + [ChannelStatus(decoded='8100'), bearer_desc_ie, buffer_size_ie]
+ return self._bip_response_head(pcmd) + [
+ ChannelStatus(decoded=self._channel_status(channel.chan_nr)),
+ bearer_desc_ie, buffer_size_ie]
def handle_CloseChannel(self, pcmd: ProactiveCommand):
"""Close a channel."""
logger.info("CloseChannel")
logger.info(pcmd)
+ dev_id_ie = Proact._find_first_element_of_type(pcmd.children, DeviceIdentities)
+ chan = self._channel_for(dev_id_ie)
+ if chan is None:
+ # channel closed / invalid
+ return self._bip_response_head(pcmd, 'bearer_independent_protocol_error',
+ 'channel_id_not_valid')
+ chan.close()
+ return self._bip_response_head(pcmd)
def handle_ReceiveData(self, pcmd: ProactiveCommand):
"""Receive/read data from the socket."""
@@ -175,6 +396,21 @@
# ]}
logger.info("ReceiveData")
logger.info(pcmd)
+ dev_id_ie = Proact._find_first_element_of_type(pcmd.children, DeviceIdentities)
+ req_len_ie = Proact._find_first_element_of_type(pcmd.children, ChannelDataLength)
+ chan = self._channel_for(dev_id_ie)
+ if chan is None:
+ return self._bip_response_head(pcmd, 'bearer_independent_protocol_error',
+ 'channel_id_not_valid')
+ # TS 102 223 8.54: RECEIVE DATA contains the requested count the card wants
+ requested = req_len_ie.decoded if req_len_ie is not None else chan.rx_available()
+ data, remaining = chan.take_rx(requested)
+ # TS 102 223 6.4.29:
+ # - return data available in the Rx buffer + num bytes still remaining (FF if > 255)
+ # - if fewer than requested available terminal must NOT wait, report and returns what we have
+ general_result = 'performed_successfully'
+ if len(data) < requested:
+ general_result = 'performed_with_missing_information'
# Terminal Response example: [
# {'command_details': {'command_number': 1,
# 'type_of_command': 'receive_data',
@@ -184,7 +420,9 @@
# {'channel_data': '16030100040e000000'},
# {'channel_data_length': 0}
# ]
- return self.prepare_response(pcmd) + []
+ return self._bip_response_head(pcmd, general_result) + [
+ ChannelData(decoded=b2h(data)),
+ ChannelDataLength(decoded=min(remaining, 0xff))]
def handle_SendData(self, pcmd: ProactiveCommand):
"""Send/write data received from the SIM to the socket."""
@@ -199,10 +437,18 @@
logger.info(pcmd)
dev_id_ie = Proact._find_first_element_of_type(pcmd.children, DeviceIdentities)
chan_data_ie = Proact._find_first_element_of_type(pcmd.children, ChannelData)
- chan_str = dev_id_ie.decoded['dest_dev_id']
- chan_nr = 1 # FIXME
- chan = self.channels.channels.get(chan_nr, None)
- # FIXME chan.prot.transport.write(h2b(chan_data_ie.decoded))
+ chan = self._channel_for(dev_id_ie)
+ if chan is None:
+ return self._bip_response_head(pcmd, 'bearer_independent_protocol_error',
+ 'channel_id_not_valid')
+ # lets accept hexstrings as well
+ payload = chan_data_ie.decoded
+ if isinstance(payload, str):
+ payload = h2b(payload)
+ # command_qualifier bit 1 selects 'send immediately' / Tx-buffer store and forward
+ # For TCP stream all we have is a socket and TCP takes care of segmentation,
+ # so just send.
+ chan.send(payload)
# Terminal Response example: [
# {'command_details': {'command_number': 1,
# 'type_of_command': 'send_data',
@@ -211,7 +457,8 @@
# {'result': {'general_result': 'performed_successfully', 'additional_information': ''}},
# {'channel_data_length': 255}
# ]
- return self.prepare_response(pcmd) + [ChannelDataLength(decoded=255)]
+ # TS 102 223 6.4.30 / 8.54 Channel data length = free space tx buf; FF == > 255 available
+ return self._bip_response_head(pcmd) + [ChannelDataLength(decoded=255)]
def handle_SetUpEventList(self, pcmd: ProactiveCommand):
# {'set_up_event_list': [{'command_details': {'command_number': 1,
diff --git a/tests/unittests/test_bip_relay.py b/tests/unittests/test_bip_relay.py
index e2e7ff8..cf11b7f 100644
--- a/tests/unittests/test_bip_relay.py
+++ b/tests/unittests/test_bip_relay.py
@@ -19,8 +19,241 @@
import unittest
-from pySim.bip import Proact
+from osmocom.utils import b2h, h2b
+
from pySim.sms import SMS_SUBMIT, AddressField
+from pySim.cat import (ProactiveCommand, CommandDetails, DeviceIdentities,
+ BearerDescription, BufferSize, UiccTransportLevel,
+ OtherAddress, ChannelData, ChannelDataLength, ChannelStatus,
+ Result)
+
+from pySim.bip import Proact
+
+
+class _EchoServer:
+ """behold, my tiny threaded TCP echo server listening on 127.0.0.1:<port>"""
+ def __init__(self):
+ self._srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ self._srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ self._srv.bind(('127.0.0.1', 0))
+ self._srv.listen(1)
+ self.port = self._srv.getsockname()[1]
+ self.accepted = threading.Event()
+ self._conns = []
+ self._stop = False
+ threading.Thread(target=self._run, daemon=True).start()
+
+ def _run(self):
+ try:
+ conn, _ = self._srv.accept()
+ except OSError:
+ return
+ self._conns.append(conn)
+ self.accepted.set()
+ while not self._stop:
+ try:
+ data = conn.recv(4096)
+ except OSError:
+ break
+ if not data:
+ break
+ conn.sendall(data)
+
+ def close(self):
+ self._stop = True
+ for s in [self._srv] + self._conns:
+ try:
+ s.close()
+ except OSError:
+ pass
+
+
+def _pcmd(children_tlvs):
+ """Assemble D0 proactive-command TLV from child IE bytes,
+ decode it like transport does after a FETCH"""
+ body = b''.join(children_tlvs)
+ pdu = h2b('D0') + bytes([len(body)]) + body
+ return ProactiveCommand().from_tlv(pdu)
+
+
+def _open_channel(port, ip='127.0.0.1', cmd_nr=1):
+ a, b, c, d = (int(x) for x in ip.split('.'))
+ return _pcmd([
+ CommandDetails(decoded={'command_number': cmd_nr, 'type_of_command': 'open_channel',
+ 'command_qualifier': 3}).to_tlv(),
+ DeviceIdentities(decoded={'source_dev_id': 'uicc', 'dest_dev_id': 'terminal'}).to_tlv(),
+ BearerDescription(decoded={'bearer_type': 'default', 'bearer_parameters': b''}).to_tlv(),
+ BufferSize(decoded=1024).to_tlv(),
+ UiccTransportLevel(decoded={'protocol_type': 'tcp_uicc_client_remote',
+ 'port_number': port}).to_tlv(),
+ OtherAddress(decoded={'type_of_address': 'ipv4',
+ 'address': bytes([a, b, c, d])}).to_tlv(),
+ ])
+
+
+def _send_data(payload, chan='channel_1', cmd_nr=1):
+ return _pcmd([
+ CommandDetails(decoded={'command_number': cmd_nr, 'type_of_command': 'send_data',
+ 'command_qualifier': 1}).to_tlv(),
+ DeviceIdentities(decoded={'source_dev_id': 'uicc', 'dest_dev_id': chan}).to_tlv(),
+ ChannelData(decoded=b2h(payload)).to_tlv(),
+ ])
+
+
+def _receive_data(length, chan='channel_1', cmd_nr=1):
+ return _pcmd([
+ CommandDetails(decoded={'command_number': cmd_nr, 'type_of_command': 'receive_data',
+ 'command_qualifier': 0}).to_tlv(),
+ DeviceIdentities(decoded={'source_dev_id': 'uicc', 'dest_dev_id': chan}).to_tlv(),
+ ChannelDataLength(decoded=length).to_tlv(),
+ ])
+
+
+def _close_channel(chan='channel_1', cmd_nr=1):
+ return _pcmd([
+ CommandDetails(decoded={'command_number': cmd_nr, 'type_of_command': 'close_channel',
+ 'command_qualifier': 0}).to_tlv(),
+ DeviceIdentities(decoded={'source_dev_id': 'uicc', 'dest_dev_id': chan}).to_tlv(),
+ ])
+
+
+def _first(til, cls):
+ return next((x for x in til if isinstance(x, cls)), None)
+
+
+class BipRelayRoundTripTest(unittest.TestCase):
+ """Drive the fixed Proact handlers (blocking sockets) with synthetic
+ proactive commands against a local echo server and assert a byte round-trip
+ plus the channel bookkeeping / error handling."""
+
+ def setUp(self):
+ self.echo = _EchoServer()
+ self.addCleanup(self.echo.close)
+ self.events = []
+ self.proact = Proact(data_available_sink=self.events.append)
+ self.addCleanup(self._close_all_channels)
+
+ def _close_all_channels(self):
+ for chan in list(self.proact.channels.channels.values()):
+ try:
+ chan.close()
+ except Exception:
+ pass
+
+ def _open(self, cmd_nr=1):
+ til = self.proact.handle_OpenChannel(_open_channel(self.echo.port, cmd_nr=cmd_nr))
+ # every TLV in the response must serialise (the transport does exactly
+ # this to post the TERMINAL RESPONSE)
+ b''.join(x.to_tlv() for x in til)
+ return til
+
+ def test_open_send_receive_roundtrip(self):
+ # OPEN CHANNEL -> socket connected, channel 1 opened, link established
+ til = self._open()
+ self.assertTrue(self.echo.accepted.wait(timeout=2.0))
+ self.assertIn(1, self.proact.channels.channels)
+ cd = _first(til, CommandDetails)
+ self.assertEqual(cd.decoded['type_of_command'], 'open_channel')
+ # TS 102 223 6.8.2 TERMINAL RESPONSE device id: terminal -> UICC
+ self.assertEqual(b2h(_first(til, DeviceIdentities).to_tlv()), '82028281')
+ # channel status: channel 1, link established
+ self.assertEqual(_first(til, ChannelStatus).decoded, '8100')
+ self.assertEqual(_first(til, Result).decoded['general_result'], 'performed_successfully')
+
+ # SEND DATA -> bytes written to the socket, echo server sends them back
+ payload = b'Hello SCP81 relay - opaque TLS record bytes'
+ til = self.proact.handle_SendData(_send_data(payload))
+ b''.join(x.to_tlv() for x in til)
+ # channel data length in the response = free Tx space, FF = ">255"
+ self.assertEqual(_first(til, ChannelDataLength).decoded, 255)
+ self.assertEqual(_first(til, Result).decoded['general_result'], 'performed_successfully')
+
+ # RECEIVE DATA -> drain the bytes back to the "card". real card
+ # uses data-available event, we poll the buffer
+ # and may need several RECEIVE DATA commands, as the spec allows.
+ got = bytearray()
+ deadline = time.monotonic() + 3.0
+ while len(got) < len(payload) and time.monotonic() < deadline:
+ chan = self.proact.channels.channels[1]
+ chan.wait_rx(1.0)
+ til = self.proact.handle_ReceiveData(_receive_data(len(payload) - len(got)))
+ b''.join(x.to_tlv() for x in til)
+ self.assertEqual(b2h(_first(til, DeviceIdentities).to_tlv()), '82028281')
+ got += h2b(_first(til, ChannelData).decoded)
+ self.assertEqual(bytes(got), payload, "byte round-trip through the BIP relay")
+
+ # CLOSE CHANNEL -> socket closed, bookkeeping cleared
+ til = self.proact.handle_CloseChannel(_close_channel())
+ b''.join(x.to_tlv() for x in til)
+ self.assertEqual(_first(til, Result).decoded['general_result'], 'performed_successfully')
+ self.assertNotIn(1, self.proact.channels.channels)
+
+ def test_data_available_event_envelope(self):
+ # The empty->non-empty Rx transition raises ENVELOPE EVENT DOWNLOAD
+ self._open()
+ self.assertTrue(self.echo.accepted.wait(timeout=2.0))
+ payload = b'PONG'
+ self.proact.handle_SendData(_send_data(payload))
+ chan = self.proact.channels.channels[1]
+ self.assertGreater(chan.wait_rx(2.0), 0)
+ # give the reader thread a beat to invoke the sink
+ deadline = time.monotonic() + 2.0
+ while not self.events and time.monotonic() < deadline:
+ time.sleep(0.01)
+ self.assertEqual(len(self.events), 1, "one data-available event on the empty->non-empty edge")
+ env = h2b(self.events[0])
+ # d6 0e | 99 01 09 (event: data available) | 82 02 82 81 terminal->UICC
+ # | b8 02 81 00 (channel 1 established) | b7 01 XX bytes available
+ self.assertEqual(b2h(env[:15]), 'd60e99010982028281b8028100b701')
+ self.assertGreaterEqual(env[15], 1)
+ self.assertLessEqual(env[15], len(payload))
+
+ def test_channel_number_from_device_identities(self):
+ # Two channels, not the old hardcoded 1
+ e2 = _EchoServer()
+ self.addCleanup(e2.close)
+ self.proact.handle_OpenChannel(_open_channel(self.echo.port))
+ # open a second channel with a second echo server
+ til2 = self.proact.handle_OpenChannel(_open_channel(e2.port))
+ self.assertEqual(sorted(self.proact.channels.channels), [1, 2])
+ self.assertEqual(_first(til2, ChannelStatus).decoded, '8200') # channel 2, established
+
+ # SEND DATA addressed to channel_2 must reach the second socket
+ self.assertTrue(e2.accepted.wait(timeout=2.0))
+ self.proact.handle_SendData(_send_data(b'two', chan='channel_2'))
+ chan2 = self.proact.channels.channels[2]
+ self.assertGreater(chan2.wait_rx(2.0), 0)
+ til = self.proact.handle_ReceiveData(_receive_data(3, chan='channel_2'))
+ self.assertEqual(h2b(_first(til, ChannelData).decoded), b'two')
+ # ..and nothing on chan 1
+ self.assertEqual(self.proact.channels.channels[1].rx_available(), 0)
+
+ def test_commands_on_closed_channel_report_bip_error(self):
+ # SEND/RECEIVE/CLOSE on a channel that was never opened must be rejected
+ # with a BIP error
+ for til in (self.proact.handle_SendData(_send_data(b'x', chan='channel_4')),
+ self.proact.handle_ReceiveData(_receive_data(1, chan='channel_4')),
+ self.proact.handle_CloseChannel(_close_channel(chan='channel_4'))):
+ b''.join(x.to_tlv() for x in til)
+ res = _first(til, Result).decoded
+ self.assertEqual(res['general_result'], 'bearer_independent_protocol_error')
+ self.assertEqual(res['additional_information'], 'channel_id_not_valid')
+
+ def test_receive_more_than_available_is_missing_info(self):
+ # terminal must NOT wait if fewer than the requested bytes are buffered,
+ # eturns what it has with "performed with missing information".
+ self._open()
+ self.assertTrue(self.echo.accepted.wait(timeout=2.0))
+ til = self.proact.handle_ReceiveData(_receive_data(10))
+ b''.join(x.to_tlv() for x in til)
+ self.assertEqual(_first(til, Result).decoded['general_result'],
+ 'performed_with_missing_information')
+ self.assertEqual(h2b(_first(til, ChannelData).decoded), b'')
+ self.assertEqual(_first(til, ChannelDataLength).decoded, 0)
+
+
+if __name__ == "__main__":
+ unittest.main()
class BipSinkTest(unittest.TestCase):
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/43545?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: If96c768f2e35c20ea3753e601059410121517b60
Gerrit-Change-Number: 43545
Gerrit-PatchSet: 1
Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>
1
0
[S] Change in pysim[master]: osmo-smdpp: fix Twisted ALPN issues with pyOpenSSL
by Hoernchen 08 Sep '26
by Hoernchen 08 Sep '26
08 Sep '26
Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/43548?usp=email )
Change subject: osmo-smdpp: fix Twisted ALPN issues with pyOpenSSL
......................................................................
osmo-smdpp: fix Twisted ALPN issues with pyOpenSSL
pyOpenSSL >= 25.0.0 makes a Context immutable once it has been used and
raises. Downgrading pyOpenSSL is not a fix either: < 25 does not import
against recent cryptography.
This server only speaks HTTP/1.1 anway so ALPN negotiation is unused
and this can be hotpatched for affected versions.
Change-Id: I5d53216f24a20625d12f0757015c19fe341303b9
---
M osmo-smdpp.py
1 file changed, 46 insertions(+), 0 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/48/43548/1
diff --git a/osmo-smdpp.py b/osmo-smdpp.py
index 2a8e478..8120b0f 100755
--- a/osmo-smdpp.py
+++ b/osmo-smdpp.py
@@ -136,6 +136,52 @@
import logging # noqa: E402
logger = logging.getLogger(__name__)
+
+def _disable_twisted_alpn_if_incompatible():
+ """Twisted <-> pyOpenSSL TLS compatibility guard applied at import.
+
+ Twisted TLSMemoryBIOFactory applies ALPN by setting the 'select' callback
+ on the SSL Context after it has already created a Connection from that
+ Context (_createConnection -> _applyProtocolNegotiation).
+ pyOpenSSL >= 25.0.0 makes a Context immutable once it has been used and
+ raises, which aborts every inbound TLS handshake, client sees unexpected-EOF
+ / decode_error that looks like a cert/cipher problem but is not.
+ pyOpenSSL < 25 does not import against recent cryptography, so downgrading
+ it is not a fix.
+
+ This server only speaks HTTP/1.1 anyway, so ALPN negotiation is not
+ needed.
+ """
+ def _major(v):
+ import re
+ m = re.match(r'\d+', (v or '').strip())
+ return int(m.group()) if m else 0
+
+ try:
+ import OpenSSL
+ except Exception:
+ return # no pyOpenSSL ???
+ pyossl_ver = getattr(OpenSSL, '__version__', '0')
+ if _major(pyossl_ver) < 25:
+ return # pre-25 pyOpenSSL allows mutating a used Context
+
+ try:
+ import twisted
+ from twisted.protocols import tls
+ except Exception:
+ return
+ factory = getattr(tls, 'TLSMemoryBIOFactory', None)
+ if factory is None or not hasattr(factory, '_applyProtocolNegotiation'):
+ return # Twisted already fixed
+
+ factory._applyProtocolNegotiation = lambda self, connection: None
+ logger.warning("Disabled Twisted ALPN negotiation: Twisted %s + "
+ "pyOpenSSL %s are incompatible for it",
+ getattr(twisted, '__version__', '?'), pyossl_ver)
+
+
+_disable_twisted_alpn_if_incompatible()
+
# HACK: make this configurable
DATA_DIR = './smdpp-data'
HOSTNAME = 'testsmdpplus1.example.com' # must match certificates!
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/43548?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: I5d53216f24a20625d12f0757015c19fe341303b9
Gerrit-Change-Number: 43548
Gerrit-PatchSet: 1
Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>
1
0