OTA: Return both packet and decoded at decryption; do not attempt to decode absent data

Change-Id: I1c488a19ad75114255c8ff1c59d903388451b37a

README.md

@@ -23,10 +23,10 @@ Git Repository
 
 You can clone from the official Osmocom  git repository using
 ```
-git clone https://gitea.osmocom.org/sim-card/pysim.git
+git clone git://git.osmocom.org/pysim.git
 ```
 
-There is a web interface at <https://gitea.osmocom.org/sim-card/pysim>.
+There is a cgit interface at <https://git.osmocom.org/pysim>
 
 
 Installation
@@ -35,26 +35,18 @@ Installation
 Please install the following dependencies:
 
  - pyscard
- - pyserial
+ - serial
  - pytlv
  - cmd2 >= 1.3.0 but < 2.0.0
  - jsonpath-ng
- - construct >= 2.9.51
+ - construct
  - bidict
  - gsm0338
- - pyyaml >= 5.1
- - termcolor
- - colorlog
 
 Example for Debian:
-```sh
-sudo apt-get install --no-install-recommends \
-	pcscd libpcsclite-dev \
-	python3 \
-	python3-setuptools \
-	python3-pyscard \
-	python3-pip
-pip3 install --user -r requirements.txt
+```
+apt-get install python3-pyscard python3-serial python3-pip python3-yaml python3-termcolor python3-colorlog
+pip3 install -r requirements.txt
 ```
 
 After installing all dependencies, the pySim applications ``pySim-read.py``, ``pySim-prog.py`` and ``pySim-shell.py`` may be started directly from the cloned repository.

contrib/jenkins.sh

@@ -1,4 +1,4 @@
-#!/bin/sh -xe
+#!/bin/sh
 # jenkins build helper script for pysim.  This is how we build on jenkins.osmocom.org
 #
 # environment variables:
@@ -6,6 +6,8 @@
 # * PUBLISH: upload manuals after building if set to "1" (ignored without WITH_MANUALS = "1")
 #
 
+set -e
+
 if [ ! -d "./pysim-testdata/" ] ; then
 	echo "###############################################"
 	echo "Please call from pySim-prog top directory"
@@ -15,7 +17,16 @@ fi
 
 virtualenv -p python3 venv --system-site-packages
 . venv/bin/activate
-pip install -r requirements.txt
+pip install pytlv
+pip install 'pyyaml>=5.1'
+pip install cmd2==1.5
+pip install jsonpath-ng
+pip install construct
+pip install bidict
+pip install gsm0338
+pip install termcolor
+pip install colorlog
+pip install pycryptodome
 
 # Execute automatically discovered unit tests first
 python -m unittest discover -v -s tests/
@@ -26,8 +37,8 @@ python -m unittest discover -v -s tests/
 # Ignore E0401: import-error
 #   pySim/utils.py:276: E0401: Unable to import 'Crypto.Cipher' (import-error)
 #   pySim/utils.py:277: E0401: Unable to import 'Crypto.Util.strxor' (import-error)
-pip install pylint==2.14.5 # FIXME: 2.15 is crashing, see OS#5668
-python -m pylint -j0 --errors-only \
+pip install pylint
+python -m pylint --errors-only \
 	--disable E1102 \
 	--disable E0401 \
 	--enable W0301 \

pySim-shell.py

@@ -23,7 +23,7 @@ import json
 import traceback
 
 import cmd2
-from cmd2 import style, Fg
+from cmd2 import style, fg
 from cmd2 import CommandSet, with_default_category, with_argparser
 import argparse
 
@@ -134,8 +134,8 @@ class PysimApp(cmd2.Cmd):
 
     def __init__(self, card, rs, sl, ch, script=None):
         super().__init__(persistent_history_file='~/.pysim_shell_history', allow_cli_args=False,
-                         auto_load_commands=False, startup_script=script)
-        self.intro = style('Welcome to pySim-shell!', fg=Fg.RED)
+                         use_ipython=True, auto_load_commands=False, startup_script=script)
+        self.intro = style('Welcome to pySim-shell!', fg=fg.red)
         self.default_category = 'pySim-shell built-in commands'
         self.card = None
         self.rs = None
@@ -145,16 +145,16 @@ class PysimApp(cmd2.Cmd):
         self.ch = ch
 
         self.numeric_path = False
-        self.add_settable(cmd2.Settable('numeric_path', bool, 'Print File IDs instead of names', self,
+        self.add_settable(cmd2.Settable('numeric_path', bool, 'Print File IDs instead of names',
                                         onchange_cb=self._onchange_numeric_path))
         self.conserve_write = True
-        self.add_settable(cmd2.Settable('conserve_write', bool, 'Read and compare before write', self,
+        self.add_settable(cmd2.Settable('conserve_write', bool, 'Read and compare before write',
                                         onchange_cb=self._onchange_conserve_write))
         self.json_pretty_print = True
         self.add_settable(cmd2.Settable('json_pretty_print',
-                          bool, 'Pretty-Print JSON output', self))
+                          bool, 'Pretty-Print JSON output'))
         self.apdu_trace = False
-        self.add_settable(cmd2.Settable('apdu_trace', bool, 'Trace and display APDUs exchanged with card', self,
+        self.add_settable(cmd2.Settable('apdu_trace', bool, 'Trace and display APDUs exchanged with card',
                                         onchange_cb=self._onchange_apdu_trace))
 
         self.equip(card, rs)
@@ -287,23 +287,23 @@ class PysimApp(cmd2.Cmd):
             sys.stderr = self._stderr_backup
 
     def _show_failure_sign(self):
-        self.poutput(style("  +-------------+", fg=Fg.LIGHT_RED))
-        self.poutput(style("  +   ##   ##   +", fg=Fg.LIGHT_RED))
-        self.poutput(style("  +    ## ##    +", fg=Fg.LIGHT_RED))
-        self.poutput(style("  +     ###     +", fg=Fg.LIGHT_RED))
-        self.poutput(style("  +    ## ##    +", fg=Fg.LIGHT_RED))
-        self.poutput(style("  +   ##   ##   +", fg=Fg.LIGHT_RED))
-        self.poutput(style("  +-------------+", fg=Fg.LIGHT_RED))
+        self.poutput(style("  +-------------+", fg=fg.bright_red))
+        self.poutput(style("  +   ##   ##   +", fg=fg.bright_red))
+        self.poutput(style("  +    ## ##    +", fg=fg.bright_red))
+        self.poutput(style("  +     ###     +", fg=fg.bright_red))
+        self.poutput(style("  +    ## ##    +", fg=fg.bright_red))
+        self.poutput(style("  +   ##   ##   +", fg=fg.bright_red))
+        self.poutput(style("  +-------------+", fg=fg.bright_red))
         self.poutput("")
 
     def _show_success_sign(self):
-        self.poutput(style("  +-------------+", fg=Fg.LIGHT_GREEN))
-        self.poutput(style("  +          ## +", fg=Fg.LIGHT_GREEN))
-        self.poutput(style("  +         ##  +", fg=Fg.LIGHT_GREEN))
-        self.poutput(style("  +  #    ##    +", fg=Fg.LIGHT_GREEN))
-        self.poutput(style("  +   ## #      +", fg=Fg.LIGHT_GREEN))
-        self.poutput(style("  +    ##       +", fg=Fg.LIGHT_GREEN))
-        self.poutput(style("  +-------------+", fg=Fg.LIGHT_GREEN))
+        self.poutput(style("  +-------------+", fg=fg.bright_green))
+        self.poutput(style("  +          ## +", fg=fg.bright_green))
+        self.poutput(style("  +         ##  +", fg=fg.bright_green))
+        self.poutput(style("  +  #    ##    +", fg=fg.bright_green))
+        self.poutput(style("  +   ## #      +", fg=fg.bright_green))
+        self.poutput(style("  +    ##       +", fg=fg.bright_green))
+        self.poutput(style("  +-------------+", fg=fg.bright_green))
         self.poutput("")
 
     def _process_card(self, first, script_path):

pySim/construct.py

@@ -2,7 +2,7 @@ from construct.lib.containers import Container, ListContainer
 from construct.core import EnumIntegerString
 import typing
 from construct import *
-from construct.core import evaluate, BitwisableString
+from construct.core import evaluate, bytes2integer, integer2bytes, BitwisableString
 from construct.lib import integertypes
 from pySim.utils import b2h, h2b, swap_nibbles
 import gsm0338
@@ -219,7 +219,7 @@ class GreedyInteger(Construct):
         if evaluate(self.swapped, context):
             data = swapbytes(data)
         try:
-            return int.from_bytes(data, byteorder='big', signed=self.signed)
+            return bytes2integer(data, self.signed)
         except ValueError as e:
             raise IntegerError(str(e), path=path)
 
@@ -248,7 +248,7 @@ class GreedyInteger(Construct):
             raise IntegerError(f"value {obj} is not an integer", path=path)
         length = self.__bytes_required(obj, self.minlen)
         try:
-            data = obj.to_bytes(length, byteorder='big', signed=self.signed)
+            data = integer2bytes(obj, length, self.signed)
         except ValueError as e:
             raise IntegerError(str(e), path=path)
         if evaluate(self.swapped, context):

pySim/ota.py

@@ -0,0 +1,461 @@
+"""Code related to SIM/UICC OTA according to TS 102 225 + TS 31.115."""
+
+# (C) 2021-2022 by Harald Welte <laforge@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/>.
+
+from pySim.construct import *
+from pySim.utils import b2h
+from pySim.sms import UserDataHeader
+from construct import *
+from bidict import bidict
+import zlib
+import abc
+import struct
+from typing import Optional
+
+# ETS TS 102 225 gives the general command structure and the dialects for CAT_TP, TCP/IP and HTTPS
+# 3GPP TS 31.115 gives the dialects for SMS-PP, SMS-CB, USSD and HTTP
+
+# CPI CPL CHI CHL SPI KIc KID TAR CNTR PCNTR RC/CC/DS data
+
+#                                           CAT_TP  TCP/IP          SMS
+# CPI                                       0x01    0x01            =IEIa=70,len=0
+# CHI                                       NULL    NULL            NULL
+# CPI, CPL and CHL included in RC/CC/DS     true    true
+# RPI                                       0x02    0x02            =IEIa=71,len=0
+# RHI                                       NULL    NULL
+# RPI, RPL and RHL included in RC/CC/DS     true    true
+# packet-id                             0-bf,ff     0-bf,ff
+# identification packet                     false   102 225 tbl 6
+
+# KVN 1..f; KI1=KIc, KI2=KID, KI3=DEK
+
+# TS 102 225 Table 5
+ota_status_codes = bidict({
+    0x00:   'PoR OK',
+    0x01:   'RC/CC/DS failed',
+    0x02:   'CNTR low',
+    0x03:   'CNTR high',
+    0x04:   'CNTR blocked',
+    0x05:   'Ciphering error',
+    0x06:   'Unidentified security error',
+    0x07:   'Insufficient memory',
+    0x08:   'more time',
+    0x09:   'TAR unknown',
+    0x0a:   'Insufficient security level',
+    0x0b:   'Actual Response in SMS-SUBMIT',    # 31.115
+    0x0c:   'Actual Response in USSD',          # 31.115
+})
+
+# ETSI TS 102 225 Table 5 + 3GPP TS 31.115 Section 7
+ResponseStatus = Enum(Int8ub, por_ok=0, rc_cc_ds_failed=1, cntr_low=2, cntr_high=3,
+                      cntr_blocked=4, ciphering_error=5, undefined_security_error=6,
+                      insufficient_memory=7, more_time_needed=8, tar_unknown=9,
+                      insufficient_security_level=0x0A,
+                      actual_response_sms_submit=0x0B,
+                      actual_response_ussd=0x0C)
+
+# ETSI TS 102 226 Section 5.1.2
+CompactRemoteResp = Struct('number_of_commands'/Int8ub,
+                           'last_status_word'/HexAdapter(Bytes(2)),
+                           'last_response_data'/HexAdapter(GreedyBytes))
+
+RC_CC_DS = Enum(BitsInteger(2), no_rc_cc_ds=0, rc=1, cc=2, ds=3)
+
+# TS 102 225 Section 5.1.1 + TS 31.115 Section 4.2
+SPI = BitStruct(  # first octet
+    Padding(3),
+    'counter'/Enum(BitsInteger(2), no_counter=0, counter_no_replay_or_seq=1,
+                   counter_must_be_higher=2, counter_must_be_lower=3),
+    'ciphering'/Flag,
+    'rc_cc_ds'/RC_CC_DS,
+    # second octet
+    Padding(2),
+    'por_in_submit'/Flag,
+    'por_shall_be_ciphered'/Flag,
+    'por_rc_cc_ds'/RC_CC_DS,
+    'por'/Enum(BitsInteger(2), no_por=0,
+               por_required=1, por_only_when_error=2)
+)
+
+# TS 102 225 Section 5.1.2
+KIC = BitStruct('key'/BitsInteger(4),
+                'algo'/Enum(BitsInteger(4), implicit=0, single_des=1, triple_des_cbc2=5, triple_des_cbc3=9,
+                            aes_cbc=2)
+                )
+
+# TS 102 225 Section 5.1.3.1
+KID_CC = BitStruct('key'/BitsInteger(4),
+                   'algo'/Enum(BitsInteger(4), implicit=0, single_des=1, triple_des_cbc2=5, triple_des_cbc3=9,
+                               aes_cmac=2)
+                   )
+
+# TS 102 225 Section 5.1.3.2
+KID_RC = BitStruct('key'/BitsInteger(4),
+                   'algo'/Enum(BitsInteger(4), implicit=0, crc16=1, crc32=5, proprietary=3)
+                   )
+
+SmsCommandPacket = Struct('cmd_pkt_len'/Int16ub,
+                          'cmd_hdr_len'/Int8ub,
+                          'spi'/SPI,
+                          'kic'/KIC,
+                          'kid'/Switch(this.spi.rc_cc_ds, {'cc': KID_CC, 'rc': KID_RC }),
+                          'tar'/Bytes(3),
+                          'secured_data'/GreedyBytes)
+
+class OtaKeyset:
+    """The OTA related data (key material, counter) to be used in encrypt/decrypt."""
+    def __init__(self, algo_crypt: str, kic_idx: int, kic: bytes,
+                 algo_auth: str, kid_idx: int, kid: bytes, cntr: int = 0):
+        self.algo_crypt = algo_crypt
+        self.kic = bytes(kic)
+        self.kic_idx = kic_idx
+        self.algo_auth = algo_auth
+        self.kid = bytes(kid)
+        self.kid_idx = kid_idx
+        self.cntr = cntr
+
+    @property
+    def auth(self):
+        """Return an instance of the matching OtaAlgoAuth."""
+        return OtaAlgoAuth.fromKeyset(self)
+
+    @property
+    def crypt(self):
+        """Return an instance of the matching OtaAlgoCrypt."""
+        return OtaAlgoCrypt.fromKeyset(self)
+
+class OtaCheckError(Exception):
+    pass
+
+class OtaDialect(abc.ABC):
+    """Base Class for OTA dialects such as SMS, BIP, ..."""
+
+    def _compute_sig_len(self, spi:SPI):
+        if spi['rc_cc_ds'] == 'no_rc_cc_ds':
+            return 0
+        elif spi['rc_cc_ds'] == 'rc': # CRC-32
+            return 4
+        elif spi['rc_cc_ds'] == 'cc': # Cryptographic Checksum (CC)
+            # TODO: this is not entirely correct, as in AES case it could be 4 or 8
+            return 8
+        else:
+            raise ValueError("Invalid rc_cc_ds: %s" % spi['rc_cc_ds'])
+
+    @abc.abstractmethod
+    def encode_cmd(self, otak: OtaKeyset, tar: bytes, apdu: bytes) -> bytes:
+        pass
+
+    @abc.abstractmethod
+    def decode_resp(self, otak: OtaKeyset, 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.
+
+        The response packet's common characteristics are not fully determined,
+        and (so far) completely proprietary per dialect."""
+        pass
+
+
+from Crypto.Cipher import DES, DES3, AES
+from Crypto.Hash import CMAC
+
+class OtaAlgo(abc.ABC):
+    iv = b'\x00\x00\x00\x00\x00\x00\x00\x00'
+    blocksize = None
+    enum_name = None
+
+    @staticmethod
+    def _get_padding(in_len: int, multiple: int, padding: int = 0):
+        """Return padding bytes towards multiple of N."""
+        if in_len % multiple == 0:
+            return b''
+        pad_cnt = multiple - (in_len % multiple)
+        return b'\x00' * pad_cnt
+
+    @staticmethod
+    def _pad_to_multiple(indat: bytes, multiple: int, padding: int = 0):
+        """Pad input bytes to multiple of N."""
+        return indat + OtaAlgo._get_padding(len(indat), multiple, padding)
+
+    def pad_to_blocksize(self, indat: bytes, padding: int = 0):
+        """Pad the given input data to multiple of the cipher block size."""
+        return self._pad_to_multiple(indat, self.blocksize, padding)
+
+    def __init__(self, otak: OtaKeyset):
+        self.otak = otak
+
+    def __str__(self):
+        return self.__class__.__name__
+
+class OtaAlgoCrypt(OtaAlgo, abc.ABC):
+    def __init__(self, otak: OtaKeyset):
+        if self.enum_name != otak.algo_crypt:
+            raise ValueError('Cannot use algorithm %s with key for %s' % (self.enum_name, otak.algo_crypt))
+        super().__init__(otak)
+
+    def encrypt(self, data:bytes) -> bytes:
+        """Encrypt given input bytes using the key material given in constructor."""
+        padded_data = self.pad_to_blocksize(data)
+        return self._encrypt(data)
+
+    def decrypt(self, data:bytes) -> bytes:
+        """Decrypt given input bytes using the key material given in constructor."""
+        return self._decrypt(data)
+
+    @abc.abstractmethod
+    def _encrypt(self, data:bytes) -> bytes:
+        """Actual implementation, to be implemented by derived class."""
+        pass
+
+    @abc.abstractmethod
+    def _decrypt(self, data:bytes) -> bytes:
+        """Actual implementation, to be implemented by derived class."""
+        pass
+
+    @classmethod
+    def fromKeyset(cls, otak: OtaKeyset) -> 'OtaAlgoCrypt':
+        """Resolve the class for the encryption algorithm of otak and instantiate it."""
+        for subc in cls.__subclasses__():
+            if subc.enum_name == otak.algo_crypt:
+                return subc(otak)
+        raise ValueError('No implementation for crypt algorithm %s' % otak.algo_auth)
+
+class OtaAlgoAuth(OtaAlgo, abc.ABC):
+    def __init__(self, otak: OtaKeyset):
+        if self.enum_name != otak.algo_auth:
+            raise ValueError('Cannot use algorithm %s with key for %s' % (self.enum_name, otak.algo_crypt))
+        super().__init__(otak)
+
+    def sign(self, data:bytes) -> bytes:
+        """Compute the CC/CR check bytes for the input data using key material
+        given in constructor."""
+        padded_data = self.pad_to_blocksize(data)
+        sig = self._sign(padded_data)
+        return sig
+
+    def check_sig(self, data:bytes, cc_received:bytes):
+        """Compute the CC/CR check bytes for the input data and compare against cc_received."""
+        cc = self.sign(data)
+        if cc_received != cc:
+            raise OtaCheckError('Received CC (%s) != Computed CC (%s)' % (b2h(cc_received), b2h(cc)))
+
+    @abc.abstractmethod
+    def _sign(self, data:bytes) -> bytes:
+        """Actual implementation, to be implemented by derived class."""
+        pass
+
+    @classmethod
+    def fromKeyset(cls, otak: OtaKeyset) -> 'OtaAlgoAuth':
+        """Resolve the class for the authentication algorithm of otak and instantiate it."""
+        for subc in cls.__subclasses__():
+            if subc.enum_name == otak.algo_auth:
+                return subc(otak)
+        raise ValueError('No implementation for auth algorithm %s' % otak.algo_auth)
+
+class OtaAlgoCryptDES(OtaAlgoCrypt):
+    """DES is insecure.  For backwards compatibility with pre-Rel8"""
+    name = 'DES'
+    enum_name = 'single_des'
+    blocksize = 8
+    def _encrypt(self, data:bytes) -> bytes:
+        cipher = DES.new(self.otak.kic, DES.MODE_CBC, self.iv)
+        return cipher.encrypt(data)
+
+    def _decrypt(self, data:bytes) -> bytes:
+        cipher = DES.new(self.otak.kic, DES.MODE_CBC, self.iv)
+        return cipher.decrypt(data)
+
+class OtaAlgoAuthDES(OtaAlgoAuth):
+    """DES is insecure.  For backwards compatibility with pre-Rel8"""
+    name = 'DES'
+    enum_name = 'single_des'
+    blocksize = 8
+    def _sign(self, data:bytes) -> bytes:
+        cipher = DES.new(self.otak.kid, DES.MODE_CBC, self.iv)
+        ciph = cipher.encrypt(data)
+        return ciph[len(ciph) - 8:]
+
+class OtaAlgoCryptDES3(OtaAlgoCrypt):
+    name = '3DES'
+    enum_name = 'triple_des_cbc2'
+    blocksize = 8
+    def _encrypt(self, data:bytes) -> bytes:
+        cipher = DES3.new(self.otak.kic, DES3.MODE_CBC, self.iv)
+        return cipher.encrypt(data)
+
+    def _decrypt(self, data:bytes) -> bytes:
+        cipher = DES3.new(self.otak.kic, DES3.MODE_CBC, self.iv)
+        return cipher.decrypt(data)
+
+class OtaAlgoAuthDES3(OtaAlgoAuth):
+    name = '3DES'
+    enum_name = 'triple_des_cbc2'
+    blocksize = 8
+    def _sign(self, data:bytes) -> bytes:
+        cipher = DES3.new(self.otak.kid, DES3.MODE_CBC, self.iv)
+        ciph = cipher.encrypt(data)
+        return ciph[len(ciph) - 8:]
+
+class OtaAlgoCryptAES(OtaAlgoCrypt):
+    name = 'AES'
+    enum_name = 'aes_cbc'
+    blocksize = 16 # TODO: is this needed?
+    def _encrypt(self, data:bytes) -> bytes:
+        cipher = AES.new(self.otak.kic, AES.MODE_CBC, self.iv)
+        return cipher.encrypt(data)
+
+    def _decrypt(self, data:bytes) -> bytes:
+        cipher = AES.new(self.otak.kic, AES.MODE_CBC, self.iv)
+        return cipher.decrypt(data)
+
+class OtaAlgoAuthAES(OtaAlgoAuth):
+    name = 'AES'
+    enum_name = 'aes_cmac'
+    blocksize = 16 # TODO: is this needed?
+    def _sign(self, data:bytes) -> bytes:
+        cmac = CMAC.new(self.otak.kid, ciphermod=AES, mac_len=8)
+        cmac.update(data)
+        ciph = cmac.digest()
+        return ciph[len(ciph) - 8:]
+
+
+
+class OtaDialectSms(OtaDialect):
+    """OTA dialect for SMS based transport, as described in 3GPP TS 31.115."""
+    SmsResponsePacket = Struct('rpl'/Int16ub,
+                               'rhl'/Int8ub,
+                               'tar'/Bytes(3),
+                               'cntr'/Bytes(5),
+                               'pcntr'/Int8ub,
+                               'response_status'/ResponseStatus,
+                               'cc_rc'/Bytes(this.rhl-10),
+                               'secured_data'/GreedyBytes)
+
+    def encode_cmd(self, otak: OtaKeyset, tar: bytes, spi: dict, apdu: bytes) -> bytes:
+        # length of signature in octets
+        len_sig = self._compute_sig_len(spi)
+        pad_cnt = 0
+        if spi['ciphering']: # ciphering is requested
+            # append padding bytes to end up with blocksize
+            len_cipher = 6 + len_sig + len(apdu)
+            padding = otak.crypt._get_padding(len_cipher, otak.crypt.blocksize)
+            pad_cnt = len(padding)
+            apdu += padding
+
+        kic = {'key': otak.kic_idx, 'algo': otak.algo_crypt}
+        kid = {'key': otak.kid_idx, 'algo': otak.algo_auth}
+
+        # CHL = number of octets from (and including) SPI to the end of RC/CC/DS
+        # 13 == SPI(2) + KIc(1) + KId(1) + TAR(3) + CNTR(5) + PCNTR(1)
+        chl = 13 + len_sig
+
+        # CHL + SPI (+ KIC + KID)
+        c = Struct('chl'/Int8ub, 'spi'/SPI, 'kic'/KIC, 'kid'/KID_CC, 'tar'/Bytes(3))
+        part_head = c.build({'chl': chl, 'spi':spi, 'kic':kic, 'kid':kid, 'tar':tar})
+        #print("part_head: %s" % b2h(part_head))
+
+        # CNTR + PCNTR (CNTR not used)
+        part_cnt = otak.cntr.to_bytes(5, 'big') + pad_cnt.to_bytes(1, 'big')
+        #print("part_cnt: %s" % b2h(part_cnt))
+
+        envelope_data = part_head + part_cnt + apdu
+        #print("envelope_data: %s" % b2h(envelope_data))
+
+        # 2-byte CPL. CPL is part of RC/CC/CPI to end of secured data, including any padding for ciphering
+        # CPL from and including CPI to end of secured data, including any padding for ciphering
+        cpl = len(envelope_data) + len_sig
+        envelope_data = cpl.to_bytes(2, 'big') + envelope_data
+        #print("envelope_data with cpl: %s" % b2h(envelope_data))
+
+        if spi['rc_cc_ds'] == 'cc':
+            cc = otak.auth.sign(envelope_data)
+            envelope_data = part_cnt + cc + apdu
+        elif spi['rc_cc_ds'] == 'rc':
+            # CRC32
+            crc32 = zlib.crc32(envelope_data) & 0xffffffff
+            envelope_data = part_cnt + crc32.to_bytes(4, 'big') + apdu
+        elif spi['rc_cc_ds'] == 'no_rc_cc_ds':
+            envelope_data = part_cnt + apdu
+        else:
+            raise ValueError("Invalid rc_cc_ds: %s" % spi['rc_cc_ds'])
+
+        #print("envelope_data with sig: %s" % b2h(envelope_data))
+
+        # encrypt as needed
+        if spi['ciphering']: # ciphering is requested
+            ciph = otak.crypt.encrypt(envelope_data)
+            envelope_data = part_head + ciph
+            # prefix with another CPL
+            cpl = len(envelope_data)
+            envelope_data = cpl.to_bytes(2, 'big') + envelope_data
+        else:
+            envelope_data = part_head + envelope_data
+
+        #print("envelope_data: %s" % b2h(envelope_data))
+
+        return envelope_data
+
+    def decode_resp(self, otak: OtaKeyset, spi: dict, data: bytes) -> ("OtaDialectSms.SmsResponsePacket", Optional["CompactRemoteResp"]):
+        if isinstance(data, str):
+            data = h2b(data)
+        # plain-text POR:   027100000e0ab000110000000000000001612f
+        # UDHL RPI IEDLa  RPL  RHL TAR    CNTR       PCNTR STS
+        # 02   71  00     000e 0a  b00011 0000000000 00    00  01 612f
+        # POR with CC:      027100001612b000110000000000000055f47118381175fb01612f
+        # POR with CC+CIPH: 027100001c12b000119660ebdb81be189b5e4389e9e7ab2bc0954f963ad869ed7c
+        if data[0] != 0x02:
+            raise ValueError('Unexpected UDL=0x%02x' % data[0])
+        udhd, remainder = UserDataHeader.fromBytes(data)
+        if not udhd.has_ie(0x71):
+            raise ValueError('RPI 0x71 not found in UDH')
+        rph_rhl_tar = remainder[:6] # RPH+RHL+TAR; not ciphered
+        res = self.SmsResponsePacket.parse(remainder)
+
+        if spi['por_shall_be_ciphered']:
+            # decrypt
+            ciphered_part = remainder[6:]
+            deciph = otak.crypt.decrypt(ciphered_part)
+            temp_data = rph_rhl_tar + deciph
+            res = self.SmsResponsePacket.parse(temp_data)
+            # remove specified number of padding bytes, if any
+            if res['pcntr'] != 0:
+                # this conditional is needed as python [:-0] renders an empty return!
+                res['secured_data'] = res['secured_data'][:-res['pcntr']]
+            remainder = temp_data
+
+        # is there a CC/RC present?
+        len_sig = res['rhl'] - 10
+        if spi['por_rc_cc_ds'] == 'no_rc_cc_ds':
+            if len_sig:
+                raise OtaCheckError('No RC/CC/DS requested, but len_sig=%u' % len_sig)
+        elif spi['por_rc_cc_ds'] == 'cc':
+            # verify signature
+            # UDH is part of CC/RC!
+            udh = data[:3]
+            # RPL, RHL, TAR, CNTR, PCNTR and STSare part of CC/RC
+            rpl_rhl_tar_cntr_pcntr_sts = remainder[:13]
+            # remove the CC/RC bytes
+            temp_data = udh + rpl_rhl_tar_cntr_pcntr_sts + remainder[13+len_sig:]
+            cc = otak.auth.check_sig(temp_data, res['cc_rc'])
+        # TODO: CRC
+        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':
+            dec = CompactRemoteResp.parse(res['secured_data'])
+        else:
+            dec = None
+        return (res, dec)

pySim/sms.py

@@ -0,0 +1,53 @@
+"""Code related to SMS Encoding/Decoding"""
+# simplistic SMS T-PDU code, as unfortunately nobody bothered to port the python smspdu
+# module to python3, and I gave up after >= 3 hours of trying and failing to do so
+
+# (C) 2022 by Harald Welte <laforge@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 typing
+from construct import Int8ub, Bytes
+from construct import Struct, Tell, this, RepeatUntil
+
+from pySim.utils import Hexstr, h2b, b2h
+
+BytesOrHex = typing.Union[Hexstr, bytes]
+
+class UserDataHeader:
+    # a single IE in the user data header
+    ie_c = Struct('offset'/Tell, 'iei'/Int8ub, 'length'/Int8ub, 'data'/Bytes(this.length))
+    # parser for the full UDH: Length octet followed by sequence of IEs
+    _construct = Struct('udhl'/Int8ub,
+                        # FIXME: somehow the below lambda is not working, we always only get the first IE?
+                        'ies'/RepeatUntil(lambda obj,lst,ctx: ctx._io.tell() > 1+this.udhl, ie_c))
+
+    def __init__(self, ies=[]):
+        self.ies = ies
+
+    def __repr__(self) -> str:
+        return 'UDH(%r)' % self.ies
+
+    def has_ie(self, iei:int) -> bool:
+        for ie in self.ies:
+            if ie['iei'] == iei:
+                return True
+        return False
+
+    @classmethod
+    def fromBytes(cls, inb: BytesOrHex) -> typing.Tuple['UserDataHeader', bytes]:
+        if isinstance(inb, str):
+            inb = h2b(inb)
+        res = cls._construct.parse(inb)
+        return cls(res['ies']), inb[1+res['udhl']:]

pySim/transport/__init__.py

@@ -10,7 +10,7 @@ from typing import Optional, Tuple
 from pySim.exceptions import *
 from pySim.construct import filter_dict
 from pySim.utils import sw_match, b2h, h2b, i2h, Hexstr
-from pySim.cat import ProactiveCommand, CommandDetails, DeviceIdentities, Result
+from pySim.cat import ProactiveCommand
 
 #
 # Copyright (C) 2009-2010  Sylvain Munaut <tnt@246tNt.com>
@@ -42,7 +42,10 @@ class ProactiveHandler(abc.ABC):
     """Abstract base class representing the interface of some code that handles
     the proactive commands, as returned by the card in responses to the FETCH
     command."""
-    def receive_fetch_raw(self, pcmd: ProactiveCommand, parsed: Hexstr):
+    def receive_fetch_raw(self, payload: Hexstr):
+        # parse the proactive command
+        pcmd = ProactiveCommand()
+        parsed = pcmd.from_tlv(h2b(payload))
         # try to find a generic handler like handle_SendShortMessage
         handle_name = 'handle_%s' % type(parsed).__name__
         if hasattr(self, handle_name):
@@ -157,57 +160,13 @@ class LinkBase(abc.ABC):
                         sw   : string (in hex) of status word (ex. "9000")
         """
         rv = self.send_apdu(pdu)
-        last_sw = rv[1]
 
-        while sw == '9000' and sw_match(last_sw, '91xx'):
-            # It *was* successful after all -- the extra pieces FETCH handled
-            # need not concern the caller.
-            rv = (rv[0], '9000')
+        while sw == '9000' and sw_match(rv[1], '91xx'):
             # proactive sim as per TS 102 221 Setion 7.4.2
-            # TODO: Check SW manually to avoid recursing on the stack (provided this piece of code stays in this place)
-            fetch_rv = self.send_apdu_checksw('80120000' + last_sw[2:], sw)
-            # Setting this in case we later decide not to send a terminal
-            # response immediately unconditionally -- the card may still have
-            # something pending even though the last command was not processed
-            # yet.
-            last_sw = fetch_rv[1]
-            # parse the proactive command
-            pcmd = ProactiveCommand()
-            parsed = pcmd.from_tlv(h2b(fetch_rv[0]))
-            print("FETCH: %s (%s)" % (fetch_rv[0], type(parsed).__name__))
-            result = Result()
+            rv = self.send_apdu_checksw('80120000' + rv[1][2:], sw)
+            print("FETCH: %s" % rv[0])
             if self.proactive_handler:
-                # Extension point: If this does return a list of TLV objects,
-                # they could be appended after the Result; if the first is a
-                # Result, that cuold replace the one built here.
-                self.proactive_handler.receive_fetch_raw(pcmd, parsed)
-                result.from_dict({'general_result': 'performed_successfully', 'additional_information': ''})
-            else:
-                result.from_dict({'general_result': 'command_beyond_terminal_capability', 'additional_information': ''})
-
-            # Send response immediately, thus also flushing out any further
-            # proactive commands that the card already wants to send
-            #
-            # Structure as per TS 102 223 V4.4.0 Section 6.8
-
-            # The Command Details are echoed from the command that has been processed.
-            (command_details,) = [c for c in pcmd.decoded.children if isinstance(c, CommandDetails)]
-            # The Device Identities are fixed. (TS 102 223 V4.0.0 Section 6.8.2)
-            device_identities = DeviceIdentities()
-            device_identities.from_dict({'source_dev_id': 'terminal', 'dest_dev_id': 'uicc'})
-
-            # Testing hint: The value of tail does not influence the behavior
-            # of an SJA2 that sent ans SMS, so this is implemented only
-            # following TS 102 223, and not fully tested.
-            tail = command_details.to_tlv() + device_identities.to_tlv() + result.to_tlv()
-            # Testing hint: In contrast to the above, this part is positively
-            # essential to get the SJA2 to provide the later parts of a
-            # multipart SMS in response to an OTA RFM command.
-            terminal_response = '80140000' + b2h(len(tail).to_bytes(1, 'big') + tail)
-
-            terminal_response_rv = self.send_apdu(terminal_response)
-            last_sw = terminal_response_rv[1]
-
+                self.proactive_handler.receive_fetch_raw(rv[0])
         if not sw_match(rv[1], sw):
             raise SwMatchError(rv[1], sw.lower(), self.sw_interpreter)
         return rv

requirements.txt

@@ -3,9 +3,10 @@ pyserial
 pytlv
 cmd2==1.5
 jsonpath-ng
-construct>=2.9.51
+construct
 bidict
 gsm0338
 pyyaml>=5.1
 termcolor
 colorlog
+pycryptodome

setup.py

@@ -14,11 +14,12 @@ setup(
         "pytlv",
         "cmd2 >= 1.3.0, < 2.0.0",
         "jsonpath-ng",
-        "construct >= 2.9.51",
+        "construct >= 2.9",
         "bidict",
         "gsm0338",
         "termcolor",
-        "colorlog"
+        "colorlog",
+        "pycryptodome"
     ],
     scripts=[
         'pySim-prog.py',

tests/test_ota.py

@@ -0,0 +1,85 @@
+#!/usr/bin/env python3
+
+import unittest
+from pySim.utils import h2b, b2h
+from pySim.ota import *
+
+class Test_SMS_3DES(unittest.TestCase):
+    tar = h2b('b00000')
+    """Test the OtaDialectSms for 3DES algorithms."""
+    def __init__(self, foo, **kwargs):
+        super().__init__(foo, **kwargs)
+        # KIC1 + KID1 of 8988211000000467285
+        KIC1 = h2b('D0FDA31990D8D64178601317191669B4')
+        KID1 = h2b('D24EB461799C5E035C77451FD9404463')
+        KIC3 = h2b('C21DD66ACAC13CB3BC8B331B24AFB57B')
+        KID3 = h2b('12110C78E678C25408233076AA033615')
+        self.od = OtaKeyset(algo_crypt='triple_des_cbc2', kic_idx=3, kic=KIC3,
+                            algo_auth='triple_des_cbc2', kid_idx=3, kid=KID3)
+        self.dialect = OtaDialectSms()
+        self.spi_base = {
+                'counter':'no_counter',
+                'ciphering': True,
+                'rc_cc_ds': 'cc',
+                'por_in_submit':False,
+                'por': 'por_required',
+                'por_shall_be_ciphered': True,
+                'por_rc_cc_ds': 'cc',
+            }
+
+    def _check_response(self, r):
+        self.assertEqual(r['number_of_commands'], 1)
+        self.assertEqual(r['last_status_word'], '612f')
+        self.assertEqual(r['last_response_data'], u'')
+        self.assertEqual(r['response_status'], 'por_ok')
+
+    def test_resp_3des_ciphered(self):
+        spi = self.spi_base
+        spi['por_shall_be_ciphered'] = True
+        spi['por_rc_cc_ds'] = 'cc'
+        r = self.dialect.decode_resp(self.od, spi, '027100001c12b000119660ebdb81be189b5e4389e9e7ab2bc0954f963ad869ed7c')
+        self._check_response(r)
+
+    def test_resp_3des_signed(self):
+        spi = self.spi_base
+        spi['por_shall_be_ciphered'] = False
+        spi['por_rc_cc_ds'] = 'cc'
+        r = self.dialect.decode_resp(self.od, spi, '027100001612b000110000000000000055f47118381175fb01612f')
+        self._check_response(r)
+
+    def test_resp_3des_signed_err(self):
+        """Expect an OtaCheckError exception if the computed CC != received CC"""
+        spi = self.spi_base
+        spi['por_shall_be_ciphered'] = False
+        spi['por_rc_cc_ds'] = 'cc'
+        with self.assertRaises(OtaCheckError) as context:
+            r = self.dialect.decode_resp(self.od, spi, '027100001612b000110000000000000055f47118381175fb02612f')
+        self.assertTrue('!= Computed CC' in str(context.exception))
+
+    def test_resp_3des_none(self):
+        spi = self.spi_base
+        spi['por_shall_be_ciphered'] = False
+        spi['por_rc_cc_ds'] = 'no_rc_cc_ds'
+        r = self.dialect.decode_resp(self.od, spi, '027100000e0ab000110000000000000001612f')
+        self._check_response(r)
+
+    def test_cmd_3des_ciphered(self):
+        spi = self.spi_base
+        spi['ciphering'] = True
+        spi['rc_cc_ds'] = 'no_rc_cc_ds'
+        r = self.dialect.encode_cmd(self.od, self.tar, spi, h2b('00a40000023f00'))
+        self.assertEqual(b2h(r), '00180d04193535b000000c8478b552a4ffc5a8f099b83cad7123')
+
+    def test_cmd_3des_signed(self):
+        spi = self.spi_base
+        spi['ciphering'] = False
+        spi['rc_cc_ds'] = 'cc'
+        r = self.dialect.encode_cmd(self.od, self.tar, spi, h2b('00a40000023f00'))
+        self.assertEqual(b2h(r), '1502193535b00000000000000000072ea17bdb72060e00a40000023f00')
+
+    def test_cmd_3des_none(self):
+        spi = self.spi_base
+        spi['ciphering'] = False
+        spi['rc_cc_ds'] = 'no_rc_cc_ds'
+        r = self.dialect.encode_cmd(self.od, self.tar, spi, h2b('00a40000023f00'))
+        self.assertEqual(b2h(r), '0d00193535b0000000000000000000a40000023f00')