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

pySim/esim/saip/batch.py

@@ -123,6 +123,8 @@ class BatchPersonalization:
                 except Exception as e:
                     raise ValueError(f'{p.param_cls.get_name()} fed by {p.src.name}: {e}') from e
 
+            pes.rebuild_mandatory_services()
+
             yield pes
 
 

pySim/esim/saip/personalization.py

@@ -1156,7 +1156,7 @@ class MilenageRotationConstants(BinaryParam, AlgoConfig):
 
 class MilenageXoringConstants(BinaryParam, AlgoConfig):
     """XOR-ing constants c1,c2,c3,c4,c5 of Milenage, 128bit each. See 3GPP TS 35.206 Sections 2.3 + 5.3.
-    Provided as octet-string concatenation of all 5 constants. The default value by 3GPP is the concetenation
+    Provided as octet-string concatenation of all 5 constants. The default value by 3GPP is the concatenation
     of::
 
      00000000000000000000000000000000

pySim/global_platform/__init__.py

@@ -18,10 +18,12 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 
 import io
+import hashlib
 from copy import deepcopy
 from typing import Optional, List, Dict, Tuple
 from construct import Optional as COptional
 from construct import Struct, GreedyRange, FlagsEnum, Int16ub, Int24ub, Padding, Bit, Const
+from construct import Construct, stream_read, stream_write
 from Cryptodome.Random import get_random_bytes
 from Cryptodome.Cipher import DES, DES3, AES
 from osmocom.utils import *
@@ -148,6 +150,24 @@ sw_table = {
     },
 }
 
+class PutKeyLength(Construct):
+    """A length field of a PUT KEY data field, GP CardSpec v2.3.1 11.8.2.3.1
+    - all lengths ASN.1 BER-TLV (ITU-T X.690 Section 8.1.3)
+    - except that the length 128 may also be coded on one byte as '80' for backwards compatibility
+    80 does not introduce the indefinite form here which is unused in GP as far as i know.
+    That legacy form is accepted when parsing, but never generated, which agrees with the spec"""
+    def _parse(self, stream, context, path):
+        first = stream_read(stream, 1, path)[0]
+        if first <= 0x80:
+            return first
+        return int.from_bytes(stream_read(stream, first & 0x7f, path), 'big')
+
+    def _build(self, obj, stream, context, path):
+        data = bertlv_encode_len(obj)
+        stream_write(stream, data, len(data), path)
+        return obj
+
+
 # GlobalPlatform 2.1.1 Section 9.1.6
 KeyType = Enum(Byte,    des=0x80,
                         tls_psk=0x85,                           # v2.3.1 Section 11.1.8
@@ -602,8 +622,8 @@ class ADF_SD(CardADF):
             See GlobalPlatform CardSpecification v2.3 Section 11.8 for details.
 
             The KCV (Key Check Values) can either be explicitly specified using `--key-check`, or will
-            otherwise be automatically generated for DES and AES keys.  You can suppress the latter using
-            `--suppress-key-check`.
+            otherwise be automatically generated for DES, AES and TLS-PSK keys.  You can suppress the
+            latter using `--suppress-key-check`.
 
             Example (SCP80 KIC/KID/KIK):
                 put_key --key-version-nr 1 --key-id 0x01    --key-type aes --key-data 000102030405060708090a0b0c0d0e0f
@@ -620,33 +640,81 @@ class ADF_SD(CardADF):
             kdb = []
             for i in range(0, len(opts.key_type)):
                 if opts.key_check and len(opts.key_check) > i:
-                    kcv = opts.key_check[i]
+                    kcv = h2b(opts.key_check[i])
                 elif opts.suppress_key_check:
-                    kcv = ''
+                    kcv = b''
                 else:
-                    kcv_bin = compute_kcv(opts.key_type[i], h2b(opts.key_data[i])) or b''
-                    kcv = b2h(kcv_bin)
-                if self._cmd.lchan.scc.scp:
-                    # encrypted key data with DEK of current SCP
-                    kcb = b2h(self._cmd.lchan.scc.scp.encrypt_key(h2b(opts.key_data[i])))
-                else:
-                    # (for example) during personalization, DEK might not be required)
-                    kcb = opts.key_data[i]
-                kdb.append({'key_type': opts.key_type[i], 'kcb': kcb, 'kcv': kcv})
+                    kcv = compute_kcv(opts.key_type[i], h2b(opts.key_data[i])) or b''
+                kdb.append({'key_type': opts.key_type[i], 'clear_key': h2b(opts.key_data[i]), 'kcv': kcv})
             p2 = opts.key_id
             if len(opts.key_type) > 1:
                 p2 |= 0x80
             self.put_key(opts.old_key_version_nr, opts.key_version_nr, p2, kdb)
 
-        # Table 11-68: Key Data Field - Format 1 (Basic Format)
-        KeyDataBasic = GreedyRange(Struct('key_type'/KeyType,
-                                          'kcb'/Prefixed(Int8ub, GreedyBytes),
-                                          'kcv'/Prefixed(Int8ub, GreedyBytes)))
+        # Table 11-68: Key Data Field - Format 1 (Basic Format).  The key component block length is
+        # BER-TLV coded (Section 11.8.2.3.1), the key check value length is always '00' - '7F'.
+        KeyDataBasic = Struct('key_type'/KeyType,
+                              'kcb'/Prefixed(PutKeyLength(), GreedyBytes),
+                              'kcv'/Prefixed(Int8ub, GreedyBytes))
 
-        def put_key(self, old_kvn:int, kvn: int, kid: int, key_dict: dict) -> bytes:
+        @classmethod
+        def encode_key_data_basic(cls, key_type: str, kcb: bytes, kcv: bytes) -> bytes:
+            """Generic Basic key data field, GP CardSpec v2.3 Table 11-68):
+                tag || L1 || <maybe L2> KCB || <1-byte length> KCV"""
+            return cls.KeyDataBasic.build({'key_type': key_type, 'kcb': kcb, 'kcv': kcv})
+
+        @classmethod
+        def encode_key_data_psk(cls, clear_key: bytes, ciphered_key: bytes, kcv: bytes) -> bytes:
+            """Single PSK TLS '85' key data field per GP Amendment B 1.2, 3.9.1 / Table 3-13:
+                85 | L1 | <L2>  <ciphered PSK key> | <KCV length> | <KCV>
+            - framing is like Basic Format, but the kcb is always GP CardSpec Table 11-70
+            so always with the length of the clear text key value, even without padding!
+            - 'ciphered_key' is DEK(block-padded clear key), no additional length prefix."""
+            kcb = bertlv_encode_len(len(clear_key)) + ciphered_key
+            return cls.encode_key_data_basic('tls_psk', kcb, kcv)
+
+        @classmethod
+        def build_put_key_data(cls, kvn: int, keys: List[dict], scp) -> bytes:
+            """Assemble the PUT KEY data field, mixed PSK + DES DEK is supported:
+            - new KVN followed by one key data field per key.
+            - tls_psk keys per GP Amendment B
+            - other key types generic Basic format
+            Param 'keys' is a dict:
+            - 'key_type' (str)
+            - 'clear_key' (bytes)
+            - 'kcv' (bytes / empty).
+            'scp' may be None (e.g. during personalization, when the DEK may not be required)."""
+            key_data = kvn.to_bytes(1, 'big')
+            for k in keys:
+                clear = k['clear_key']
+                if k['key_type'] == 'tls_psk':
+                    # len always part of the data see CardSpec Table 11-70 vs Table 11-71
+                    if scp:
+                        ciphered = scp.dek_encrypt(scp.pad_to_blocksize(clear))
+                    else:
+                        ciphered = clear
+                    key_data += cls.encode_key_data_psk(clear, ciphered, k['kcv'])
+                else:
+                    if scp:
+                        ciphered = scp.encrypt_key(clear)
+                    else:
+                        # (for example) during personalization, DEK might not be required
+                        ciphered = clear
+                    key_data += cls.encode_key_data_basic(k['key_type'], ciphered, k['kcv'])
+            return key_data
+
+        def put_key(self, old_kvn:int, kvn: int, kid: int, keys: List[dict]) -> bytes:
             """Perform the GlobalPlatform PUT KEY command in order to store a new key on the card.
             See GlobalPlatform CardSpecification v2.3 Section 11.8 for details."""
-            key_data = kvn.to_bytes(1, 'big') + build_construct(ADF_SD.AddlShellCommands.KeyDataBasic, key_dict)
+            key_data = self.build_put_key_data(kvn, keys, self._cmd.lchan.scc.scp)
+            # Lc of Table 11-64 is a single byte, while LOAD or STORE DATA splits we can't:
+            # 11.8.2.3.3 splits a key at component boundaries -> not helping here
+            max_cmd_len = self._cmd.lchan.scc.max_cmd_len
+            if len(key_data) > max_cmd_len:
+                raise ValueError('key data field of %u bytes exceeds the maximum command length of %u '
+                                 '(limited by the overhead of the current secure channel); use fewer '
+                                 'keys per command, a single key component that large needs STORE DATA' %
+                                 (len(key_data), max_cmd_len))
             hdr = "80D8%02x%02x%02x" % (old_kvn, kid, len(key_data))
             data, _sw = self._cmd.lchan.scc.send_apdu_checksw(hdr + b2h(key_data) + "00")
             return data
@@ -826,23 +894,32 @@ class ADF_SD(CardADF):
         load_parser_from_grp.add_argument('--from-hex', type=is_hexstr, help='load from hex string')
         load_parser_from_grp.add_argument('--from-file', type=argparse.FileType('rb', 0), help='load from binary file')
         load_parser_from_grp.add_argument('--from-cap-file', type=argparse.FileType('rb', 0), help='load from JAVA-card CAP file')
+        load_parser.add_argument('--chunk-len', type=auto_uint8, default=None,
+                                 help='Block size for the LOAD command; default: as large as the current secure channel overhead permits, at most 240')
 
         @cmd2.with_argparser(load_parser)
         def do_load(self, opts):
             """Perform a GlobalPlatform LOAD command. (We currently only support loading without DAP and
             without ciphering.)"""
             if opts.from_hex is not None:
-                self.load(h2b(opts.from_hex))
+                self.load(h2b(opts.from_hex), opts.chunk_len)
             elif opts.from_file is not None:
-                self.load(opts.from_file.read())
+                self.load(opts.from_file.read(), opts.chunk_len)
             elif opts.from_cap_file is not None:
                 cap = CapFile(opts.from_cap_file)
-                self.load(cap.get_loadfile())
+                self.load(cap.get_loadfile(), opts.chunk_len)
             else:
                 raise ValueError('load source not specified!')
 
-        def load(self, contents:bytes, chunk_len:int = 240):
-            # TODO:tune chunk_len based on the overhead of the used SCP?
+        def load(self, contents:bytes, chunk_len:Optional[int] = None):
+            # scc.max_cmd_len knows the overhead the currently active SCP
+            # 240 is the old default, keep it for now.
+            max_chunk_len = self._cmd.lchan.scc.max_cmd_len
+            if chunk_len is None:
+                chunk_len = min(240, max_chunk_len)
+            elif not 1 <= chunk_len <= max_chunk_len:
+                raise ValueError('chunk_len must be in range 1..%u (limited by the overhead of the current secure channel)' %
+                                 max_chunk_len)
             # build TLV according to GPC_SPE_034 section 11.6.2.3 / Table 11-58 for unencrypted case
             remainder = b'\xC4' + bertlv_encode_len(len(contents)) + contents
             # transfer this in various chunks to the card
@@ -881,6 +958,8 @@ class ADF_SD(CardADF):
         install_cap_parser_inst_prm_grp.add_argument('--install-parameters-stk',
                                                      type=is_hexstr, default=None,
                                                      help='Load Parameters (ETSI TS 102 226, section 8.2.1.3.2.1)')
+        install_cap_parser.add_argument('--chunk-len', type=auto_uint8, default=None,
+                                        help='Block size for the LOAD command; default: as large as the current secure channel overhead permits, at most 240')
 
         @cmd2.with_argparser(install_cap_parser)
         def do_install_cap(self, opts):
@@ -919,7 +998,7 @@ class ADF_SD(CardADF):
             self._cmd.poutput("step #1: install for load...")
             self.do_install_for_load("--load-file-aid %s --security-domain-aid %s" % (load_file_aid, security_domain_aid))
             self._cmd.poutput("step #2: load...")
-            self.load(load_file)
+            self.load(load_file, opts.chunk_len)
             self._cmd.poutput("step #3: install_for_install (and make selectable)...")
             self.do_install_for_install("--load-file-aid %s --module-aid %s --application-aid %s --install-parameters %s --make-selectable" %
                                         (load_file_aid, module_aid, application_aid, install_parameters))
@@ -1065,10 +1144,16 @@ def compute_kcv_aes(key:bytes) -> bytes:
     cipher = AES.new(key, AES.MODE_ECB)
     return cipher.encrypt(plaintext)
 
+def compute_kcv_psk(key:bytes) -> bytes:
+    # GP Amendment B v1.2, 3.9.1 / Table 3-13
+    # KCV of a PSK TLS key is the 3 highest-order bytes of the SHA-1 digest of the clear key value.
+    return hashlib.sha1(key).digest()
+
 # dict is keyed by the string name of the KeyType enum above in this file
 KCV_CALCULATOR = {
         'aes': compute_kcv_aes,
         'des': compute_kcv_des,
+        'tls_psk': compute_kcv_psk,
     }
 
 def compute_kcv(key_type: str, key: bytes) -> Optional[bytes]:

pySim/global_platform/scp.py

@@ -182,6 +182,29 @@ class SCP(SecureChannel, abc.ABC):
         """Should we perform R-ENC?"""
         return self.security_level & 0x20
 
+    @property
+    @abc.abstractmethod
+    def mac_len(self) -> int:
+        """Length of the appended C-MAC, to be provided by derived class."""
+
+    @property
+    def overhead(self) -> int:
+        """Worst-case len that wrapping a command APDU adds to its data field at the
+        current sec level is (255 - overhead), C-MAC + C-DECRYPTION encryption padding."""
+        if not self.do_cmac:
+            return 0
+        if not self.do_cenc:
+            return self.mac_len
+        # see Secure Channel Protocol '03' Card Specification v2.3 - Amendment D v1.1.2
+        # which defers to GPCS v2.3 Section B.2 which then defers to
+        # NIST SP 800-38B for encryption and points out that
+        # the padding is, as expected, just the usual padding from NIST SP 800-38A
+        # C-DECRYPTION pads with ('80'+['00'...] at least 1 byte) up to
+        # the cipher block size + C-MAC on top -> largest usable data field
+        # is one byte less than the largest block-size multiple within 255 - mac_len.
+        bs = self.sk.blocksize
+        return 255 - ((255 - self.mac_len) // bs * bs - 1)
+
     def __str__(self) -> str:
         return "%s[%02x]" % (self.__class__.__name__, self.security_level)
 
@@ -215,11 +238,20 @@ class SCP(SecureChannel, abc.ABC):
     def gen_ext_auth_apdu(self, security_level: int = 0x01) -> bytes:
         pass
 
+    def pad_to_blocksize(self, data: bytes) -> bytes:
+        """Right pad the data with zero bytes to a multiple of the DEK cipher block size."""
+        if len(data) % self.sk.blocksize:
+            # not '+=' which would mutate the callers bytearray in place..
+            data = data + b'\x00' * (self.sk.blocksize - len(data) % self.sk.blocksize)
+        return data
+
     def encrypt_key(self, key: bytes) -> bytes:
         """Encrypt a key with the DEK."""
-        num_pad = len(key) % self.sk.blocksize
-        if num_pad:
-            return bertlv_encode_len(len(key)) + self.dek_encrypt(key + b'\x00'*num_pad)
+        if len(key) % self.sk.blocksize:
+            # The kcv is right padded before encryption and the kcb
+            # is formatted as described in Table 11-70: preceded by the actual length of the
+            # clear text kcv.
+            return bertlv_encode_len(len(key)) + self.dek_encrypt(self.pad_to_blocksize(key))
         return self.dek_encrypt(key)
 
     def decrypt_key(self, encrypted_key:bytes) -> bytes:
@@ -232,9 +264,8 @@ class SCP(SecureChannel, abc.ABC):
             # Block provides the actual length of the key component value, which allows recovering the
             # clear-text key component value after decryption of the encrypted key component value and removal
             # of padding bytes.
-            decrypted = self.dek_decrypt(encrypted_key)
-            key_len, remainder = bertlv_parse_len(decrypted)
-            return remainder[:key_len]
+            key_len, remainder = bertlv_parse_len(encrypted_key)
+            return self.dek_decrypt(remainder)[:key_len]
         else:
             # If the length of the Key Component Block is a multiple of the block size of the encryption
             # algorithm (i.e.  8 bytes for DES, 16 bytes for AES), then it shall be assumed that no padding
@@ -260,10 +291,8 @@ class SCP02(SCP):
     # Key Version Number 0x70 is a non-spec special-case of sysmoISIM-SJA2/SJA5 and possibly more sysmocom products
     # Key Version Number 0x01 is a non-spec special-case of sysmoUSIM-SJS1
     kvn_ranges = [[0x01, 0x01], [0x20, 0x2f], [0x70, 0x70]]
-
-    def __init__(self, *args, **kwargs):
-        self.overhead = 8
-        super().__init__(*args, **kwargs)
+    # C-MAC (Single DES + final 3DES, B.1.2.2) is always one full DES block
+    mac_len = 8
 
     def dek_encrypt(self, plaintext:bytes) -> bytes:
         # See also GPC section B.1.1.2, E.4.7, and E.4.1
@@ -338,10 +367,16 @@ class SCP02(SCP):
             # CMAC on modified APDU
             mlc = lc + 8
             clac = cla | CLA_SM
+        if mlc >= 256:
+            raise ValueError('Modified Lc (%u) would exceed maximum when appending 8 bytes of mac' % mlc)
         mac = self.sk.calc_mac_1des(bytes([clac]) + apdu[1:4] + bytes([mlc]) + data)
         if self.do_cenc:
+            padded_data = pad80(data, 8)
+            if len(padded_data) + 8 >= 256:
+                raise ValueError('Modified Lc (%u) would exceed maximum when appending padding and mac' %
+                                 (len(padded_data) + 8))
             k = DES3.new(self.sk.enc, DES.MODE_CBC, b'\x00'*8)
-            data = k.encrypt(pad80(data, 8))
+            data = k.encrypt(padded_data)
             lc = len(data)
 
         lc += 8
@@ -477,9 +512,13 @@ class SCP03(SCP):
 
     def __init__(self, *args, **kwargs):
         self.s_mode = kwargs.pop('s_mode', 8)
-        self.overhead = self.s_mode
         super().__init__(*args, **kwargs)
 
+    @property
+    def mac_len(self) -> int:
+        # C-MAC truncated to 8 in S8 or 16 bytes in S16 mode
+        return self.s_mode
+
     def dek_encrypt(self, plaintext:bytes) -> bytes:
         cipher = AES.new(self.card_keys.dek, AES.MODE_CBC, b'\x00'*16)
         return cipher.encrypt(plaintext)

pySim/ts_102_221.py

@@ -17,6 +17,7 @@ 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 bidict import bidict
+import copy
 
 from construct import Select, Const, Bit, Struct, Int16ub, FlagsEnum, GreedyString, ValidationError
 from construct import Optional as COptional, Computed
@@ -335,6 +336,8 @@ class TerminalCapability(BER_TLV_IE, tag=0xa9, nested=[TerminalPowerSupply, Exte
 
 # ETSI TS 102 221 Section 9.2.7 + ISO7816-4 9.3.3/9.3.4
 class _AM_DO_DF(DataObject):
+    """ISO7816-4:2005 5.4.3.1 Table 16"""
+
     def __init__(self):
         super().__init__('access_mode', 'Access Mode', tag=0x80)
 
@@ -381,7 +384,7 @@ class _AM_DO_DF(DataObject):
 
 
 class _AM_DO_EF(DataObject):
-    """ISO7816-4 9.3.2 Table 18 + 9.3.3.1 Table 31"""
+    """ISO7816-4:2005 5.4.3.1 Table 17"""
 
     def __init__(self):
         super().__init__('access_mode', 'Access Mode', tag=0x80)
@@ -429,7 +432,7 @@ class _AM_DO_EF(DataObject):
 
 
 class _AM_DO_CHDR(DataObject):
-    """Command Header Access Mode DO according to ISO 7816-4 Table 32."""
+    """Command Header Access Mode DO according to ISO 7816-4:2005 5.4.3.2 Table 22."""
 
     def __init__(self, tag):
         super().__init__('command_header', 'Command Header Description', tag=tag)
@@ -543,8 +546,9 @@ class CRT_DO(DataObject):
         pin = pin_names.inverse[self.decoded]
         return b'\x83\x01' + pin.to_bytes(1, 'big') + b'\x95\x01\x08'
 
-# ISO7816-4 9.3.3 Table 33
 class SecCondByte_DO(DataObject):
+    """ISO7816-4:2005 5.4.3.1 Table 20"""
+
     def __init__(self, tag=0x9d):
         super().__init__('security_condition_byte', tag=tag)
 
@@ -732,36 +736,57 @@ class EF_ARR(LinFixedEF):
                 raise ValueError
         return by_mode
 
+    @staticmethod
+    def __get_do_sequence(decode_for_df : bool = False):
+        if decode_for_df:
+            return DataObjectSequence('arr', sequence=[AM_DO_DF, SC_DO])
+        else:
+            return DataObjectSequence('arr', sequence=[AM_DO_EF, SC_DO])
+
     def _decode_record_bin(self, raw_bin_data, **kwargs):
-        # we can only guess if we should decode for EF or DF here :(
-        arr_seq = DataObjectSequence('arr', sequence=[AM_DO_EF, SC_DO])
+        # we can only guess if we should decode for EF or DF here, but our caller may
+        # be able to pass us a hint:
+        arr_seq = self.__get_do_sequence(kwargs.get('decode_for_df', False))
         dec = arr_seq.decode_multi(raw_bin_data)
         # we cannot pass the result through flatten() here, as we don't have a related
         # 'un-flattening' decoder, and hence would be unable to encode :(
         return dec[0]
 
     def _encode_record_bin(self, in_json, **kwargs):
-        # we can only guess if we should decode for EF or DF here :(
-        arr_seq = DataObjectSequence('arr', sequence=[AM_DO_EF, SC_DO])
+        # we can only guess if we should decode for EF or DF here, but our caller may
+        # be able to pass us a hint:
+        arr_seq = self.__get_do_sequence(kwargs.get('encode_for_df', False))
         return arr_seq.encode_multi(in_json)
 
     @with_default_category('File-Specific Commands')
     class AddlShellCommands(CommandSet):
-        @cmd2.with_argparser(LinFixedEF.ShellCommands.read_rec_dec_parser)
+        read_arr_argparser = copy.deepcopy(LinFixedEF.ShellCommands.read_rec_dec_parser)
+        read_arr_argparser.add_argument('--decode-for-df', action='store_true',
+                                        help='Decode EF.ARR record as if used by a DF (default: EF)')
+
+        @cmd2.with_argparser(read_arr_argparser)
         def do_read_arr_record(self, opts):
             """Read one EF.ARR record in flattened, human-friendly form."""
-            (data, _sw) = self._cmd.lchan.read_record_dec(opts.RECORD_NR)
+            (hexdata, _sw) = self._cmd.lchan.read_record(opts.RECORD_NR)
+            data = self._cmd.lchan.selected_file._decode_record_bin(h2b(hexdata),
+                                                                    decode_for_df = opts.decode_for_df)
             data = self._cmd.lchan.selected_file.flatten(data)
             self._cmd.poutput_json(data, opts.oneline)
 
-        @cmd2.with_argparser(LinFixedEF.ShellCommands.read_recs_dec_parser)
+        read_arrs_argparser = copy.deepcopy(LinFixedEF.ShellCommands.read_recs_dec_parser)
+        read_arrs_argparser.add_argument('--decode-for-df', action='store_true',
+                                        help='Decode EF.ARR records as if used by a DF (default: EF)')
+
+        @cmd2.with_argparser(read_arrs_argparser)
         def do_read_arr_records(self, opts):
             """Read + decode all EF.ARR records in flattened, human-friendly form."""
             num_of_rec = self._cmd.lchan.selected_file_num_of_rec()
             # collect all results in list so they are rendered as JSON list when printing
             data_list = []
             for recnr in range(1, 1 + num_of_rec):
-                (data, _sw) = self._cmd.lchan.read_record_dec(recnr)
+                (hexdata, _sw) = self._cmd.lchan.read_record(recnr)
+                data = self._cmd.lchan.selected_file._decode_record_bin(h2b(hexdata),
+                                                                        decode_for_df = opts.decode_for_df)
                 data = self._cmd.lchan.selected_file.flatten(data)
                 data_list.append(data)
             self._cmd.poutput_json(data_list, opts.oneline)

pySim/ts_31_102.py

@@ -335,7 +335,7 @@ class EF_SUCI_Calc_Info(TransparentEF):
         """conversion method to generate list of {hnet_pubkey_identifier, hnet_pubkey} dicts
         from flat [{hnet_pubkey_identifier: }, {net_pubkey: }, ...] list"""
         out = []
-        while len(l):
+        while l:
             a = l.pop(0)
             b = l.pop(0)
             z = {**a, **b}

tests/unittests/test_configurable_parameters.py

@@ -55,8 +55,6 @@ class ConfigurableParameterTest(unittest.TestCase):
         upp_fnames = (
                 'TS48v5_SAIP2.1A_NoBERTLV.der',
                 'TS48v5_SAIP2.3_BERTLV_SUCI.der',
-                'TS48v5_SAIP2.1B_NoBERTLV.der',
-                'TS48v5_SAIP2.3_NoBERTLV.der',
                )
 
         class Paramtest:

tests/unittests/test_globalplatform.py

@@ -17,7 +17,10 @@
 
 import unittest
 import logging
+import hashlib
+from types import SimpleNamespace
 from osmocom.utils import b2h, h2b
+from osmocom.tlv import bertlv_encode_len
 
 from pySim.global_platform import *
 from pySim.global_platform.scp import *
@@ -283,6 +286,41 @@ class SCP03_Test_AES256_33(SCP03_Test, unittest.TestCase):
 # FIXME: test auth with random (0x60) vs pseudo-random (0x70) challenge
 
 
+class KeyComponentBlock_Test(unittest.TestCase):
+    """Tests for the kcb of GP CardSpec v2.3
+    - Table 11-70 kcv that required padding, preceded by its clear-text length
+    - Table 11-71 no padding required"""
+
+    def setUp(self):
+        # SCP02 (3DES DEK, 8 byte blocks), same vectors as SCP02_Test
+        self.scp02 = SCP02(card_keys=ck_3des_70)
+        self.scp02.gen_init_update_apdu(host_challenge=h2b('40A62C37FA6304F8'))
+        self.scp02.parse_init_update_resp(h2b('00000000000000000000700200016B4524ABEE7CF32EA3838BC148F3'))
+        self.scp02.gen_ext_auth_apdu()
+        # SCP03 (AES DEK, 16 byte blocks), same vectors as SCP03_Test_AES128_11
+        self.scp03 = SCP03(card_keys=KEYSET_AES128)
+        self.scp03.gen_init_update_apdu(h2b('b13e5f938fc108c4'))
+        self.scp03.parse_init_update_resp(h2b('000000000000000000003003703eb51047495b249f66c484c1d2ef1948000002'))
+        self.scp03.gen_ext_auth_apdu(0x11)
+
+    def test_encrypt_decrypt_key(self):
+        for scp in (self.scp02, self.scp03):
+            bs = scp.sk.blocksize
+            for keylen in range(1, 3 * bs + 1):
+                with self.subTest(scp=type(scp).__name__, keylen=keylen):
+                    key = bytes(range(keylen))
+                    kcb = scp.encrypt_key(key)
+                    if keylen % bs:
+                        # Table 11-70: <length of clear key component> || <encrypted padded value>
+                        self.assertEqual(kcb[0], keylen)
+                        self.assertEqual((len(kcb) - 1) % bs, 0)
+                        self.assertEqual(len(kcb) - 1, keylen + (bs - keylen % bs))
+                    else:
+                        # Table 11-71: only the encrypted key component value
+                        self.assertEqual(len(kcb), keylen)
+                    self.assertEqual(scp.decrypt_key(kcb), key)
+
+
 class SCP03_KCV_Test(unittest.TestCase):
     def test_kcv(self):
         self.assertEqual(compute_kcv('aes', KEYSET_AES128.enc), h2b('C35280'))
@@ -290,6 +328,208 @@ class SCP03_KCV_Test(unittest.TestCase):
         self.assertEqual(compute_kcv('aes', KEYSET_AES128.dek), h2b('840DE5'))
 
 
+class PutKey_PSK_Test(unittest.TestCase):
+    """Tests for the PUT KEY command data field encoding, in particular the PSK TLS ('85') key data
+    field defined by GlobalPlatform Amendment B (Remote Application Management over HTTP) Table 3-13."""
+
+    # the PUT KEY encoder we exercise
+    C = ADF_SD.AddlShellCommands
+
+    # SCP80 TLS-PSK example key from the do_put_key docstring (16 bytes)
+    PSK_CLEAR = h2b('303132333435363738393a3b3c3d3e3f')
+    # its DEK ciphertext + Table 3-13 KCV with SCP02 session set up below
+    PSK_CIPHERED = h2b('15abf1fe16ccc5aa13743394442942cd')
+    PSK_KCV = h2b('06125d')  # = SHA-1(PSK_CLEAR)[:3]
+
+    def setUp(self):
+        # SCP02 with the same vectors as SCP02_Test, so that the whole PUT KEY data field is reproducible.
+        self.scp02 = SCP02(card_keys=ck_3des_70)
+        self.scp02.gen_init_update_apdu(host_challenge=h2b('40A62C37FA6304F8'))
+        self.scp02.parse_init_update_resp(h2b('00000000000000000000700200016B4524ABEE7CF32EA3838BC148F3'))
+        self.scp02.gen_ext_auth_apdu()
+
+    def test_psk_kcv_is_sha1(self):
+        # GP Amendment B Table 3-13: KCV = 3 most significant bytes of SHA-1(clear key)
+        self.assertEqual(compute_kcv('tls_psk', self.PSK_CLEAR), hashlib.sha1(self.PSK_CLEAR).digest()[:3])
+        self.assertEqual(compute_kcv('tls_psk', self.PSK_CLEAR), self.PSK_KCV)
+
+    def test_encode_psk_framing_golden(self):
+        # assert the exact Table 3-13 layout
+        #   85 | L1 | L2 | <ciphered> | 03 | <SHA-1(clear)[:3]>
+        clear = self.PSK_CLEAR
+        ciphered = h2b('aabbccddeeff00112233445566778899')  # arbitrary 16-byte ciphertext
+        kcv = hashlib.sha1(clear).digest()[:3]
+        field = self.C.encode_key_data_psk(clear, ciphered, kcv)
+        #                            85   L1   L2   <---------- ciphered ----------->  03  <-kcv->
+        self.assertEqual(b2h(field),'85' '11' '10' 'aabbccddeeff00112233445566778899' '03' + b2h(kcv))
+        self.assertEqual(b2h(field),'851110aabbccddeeff0011223344556677889903' + '06125d')
+
+    def test_psk_golden_over_scp02(self):
+        # Full PUT KEY data field (KVN 0x40 + single PSK key) enciphered with the SCP02 DEK.
+        keys = [{'key_type': 'tls_psk', 'clear_key': self.PSK_CLEAR,
+                 'kcv': compute_kcv('tls_psk', self.PSK_CLEAR)}]
+        data = self.C.build_put_key_data(0x40, keys, self.scp02)
+        self.assertEqual(b2h(data),
+                         '40' '85' '11' '10' + b2h(self.PSK_CIPHERED) + '03' + b2h(self.PSK_KCV))
+
+    def test_wrong_basic_format_differs(self):
+        # regression test, the generic "Basic format" does NOT match Table 3-13 for a PSK key
+        # rejected by card with with 6a88
+        wrong_basic = self.C.encode_key_data_basic('tls_psk', self.PSK_CIPHERED, b'')
+        right_psk = self.C.encode_key_data_psk(self.PSK_CLEAR, self.PSK_CIPHERED, self.PSK_KCV)
+        self.assertEqual(b2h(wrong_basic), '8510' + b2h(self.PSK_CIPHERED) + '00')
+        self.assertEqual(b2h(right_psk), '8511' '10' + b2h(self.PSK_CIPHERED) + '03' + b2h(self.PSK_KCV))
+        self.assertNotEqual(wrong_basic, right_psk)
+
+    def test_key_component_block_length_is_bertlv(self):
+        # GP CardSpec v2.3.1 Section 11.8.2.3.1: all lengths ofPUT KEY are always BER TLV coded
+        for kcb_len, exp_len_field in [(127, '7f'), (128, '8180'), (129, '8181'), (256, '820100')]:
+            with self.subTest(kcb_len=kcb_len):
+                kcb = bytes(kcb_len)
+                field = self.C.encode_key_data_basic('rsa_modulus_n', kcb, b'')
+                self.assertEqual(b2h(field), 'a2' + exp_len_field + b2h(kcb) + '00')
+                # 85 field of Amendment B Table 3-13 uses the same coding
+                # single byte inner length (clear key < 128) == block kcb_len bytes long
+                psk = self.C.encode_key_data_psk(bytes(120), bytes(kcb_len - 1), b'')
+                self.assertEqual(b2h(psk)[:2 + len(exp_len_field)], '85' + exp_len_field)
+
+    def test_basic_format_unchanged(self):
+        # as before
+        for kt, clear in [('des', h2b('404142434445464748494a4b4c4d4e4f')),
+                          ('aes', h2b('000102030405060708090a0b0c0d0e0f'))]:
+            ciph = self.scp02.encrypt_key(clear)
+            kcv = compute_kcv(kt, clear)
+            via_construct = build_construct(self.C.KeyDataBasic, {'key_type': kt, 'kcb': b2h(ciph), 'kcv': b2h(kcv)})
+            via_helper = self.C.encode_key_data_basic(kt, ciph, kcv)
+            self.assertEqual(via_helper, via_construct)
+
+    def test_psk_padding_no_double_length(self):
+        # A PSK key whose length is not a multiple of the DEK block size (DES: 8) is right-padded before
+        # ciphering. Table 3-13 states the clear key length (L2) in the '85' DO itself, so the ciphered
+        # key field is the bare cryptogram:
+        # - ciphered field == padded ciphertext (no duplicated length prefix),
+        # - clear key == first L2 bytes.
+        for keylen in (18, 20):
+            with self.subTest(keylen=keylen):
+                clear = bytes(range(keylen))
+                padded_len = keylen + (-keylen % 8)
+                field = self.C.build_put_key_data(0x40, [{'key_type': 'tls_psk', 'clear_key': clear,
+                                                          'kcv': compute_kcv('tls_psk', clear)}], self.scp02)[1:]
+                self.assertEqual(field[0], 0x85)
+                l1 = field[1]
+                l2 = field[2]
+                self.assertEqual(l2, keylen)                # single-byte BER length of clear key
+                ciphered = field[3:3 + (l1 - 1)]            # value = L2 (1 byte) || ciphered key
+                self.assertEqual(len(ciphered), padded_len) # padded to the 8-byte DES block size
+                self.assertEqual(l1, 1 + padded_len)        # no duplicated length prefix
+                self.assertEqual(self.scp02.dek_decrypt(ciphered)[:keylen], clear)
+
+    def test_psk_clear_key_is_not_padded_in_place(self):
+        # padding the bytearray in place would make L2 the padded length,
+        # then stored as key material and rejected thanks to the KCV
+        clear = h2b('000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d')  # 30, not %8
+        kcv = compute_kcv('tls_psk', clear)
+        field = self.C.build_put_key_data(0x40, [{'key_type': 'tls_psk', 'clear_key': clear,
+                                                  'kcv': kcv}], self.scp02)[1:]
+        self.assertEqual(len(clear), 30)
+        self.assertEqual(field[2], 30)                       # L2 == clear key length, not 32
+        self.assertEqual(self.scp02.dek_decrypt(field[3:3 + field[1] - 1])[:30], clear)
+
+    def test_kcv_suppressed(self):
+        # --suppress-key-check -> KCV length 00 and no KCV bytes
+        field = self.C.build_put_key_data(0x40, [{'key_type': 'tls_psk', 'clear_key': self.PSK_CLEAR,
+                                                  'kcv': b''}], self.scp02)[1:]
+        self.assertEqual(b2h(field), '8511' '10' + b2h(self.PSK_CIPHERED) + '00')
+
+    def test_multikey_psk_plus_des_dek(self):
+        # load a PSK TLS key (KID 1, Amendment B format) together with its DES DEK
+        # (KID 2, Basic format) in one PUT KEY.
+        # Verify the concatenated data field parses back into the two components with proper type formats.
+        dek = h2b('404142434445464748494a4b4c4d4e4f')
+        keys = [{'key_type': 'tls_psk', 'clear_key': self.PSK_CLEAR, 'kcv': compute_kcv('tls_psk', self.PSK_CLEAR)},
+                {'key_type': 'des', 'clear_key': dek, 'kcv': compute_kcv('des', dek)}]
+        data = self.C.build_put_key_data(0x40, keys, self.scp02)
+
+        b = data
+        self.assertEqual(b[0], 0x40)            # KVN
+        b = b[1:]
+        # component 1: PSK TLS (Table 3-13)
+        self.assertEqual(b[0], 0x85)
+        self.assertEqual(b[1], 0x11)            # L1 = 17
+        self.assertEqual(b[2], 0x10)            # L2 = 16 (clear key length)
+        self.assertEqual(b[3:3 + 16], self.PSK_CIPHERED)
+        self.assertEqual(b[3 + 16], 0x03)       # KCV length
+        self.assertEqual(b[3 + 16 + 1:3 + 16 + 1 + 3], self.PSK_KCV)
+        b = b[3 + 16 + 1 + 3:]
+        # component 2: DES DEK (Basic format)
+        self.assertEqual(b[0], 0x80)            # key type des
+        kcb_len = b[1]
+        self.assertEqual(kcb_len, 16)
+        self.assertEqual(b[2:2 + kcb_len], self.scp02.encrypt_key(dek))
+        b = b[2 + kcb_len:]
+        self.assertEqual(b[0], 0x03)            # KCV length
+        self.assertEqual(b[1:1 + 3], compute_kcv('des', dek))
+        self.assertEqual(b[1 + 3:], b'')        # no trailing bytes
+
+    def test_no_scp_leaves_key_clear(self):
+        # During personalization (no SCP) the key is not enciphered, framing still follows Table 3-13.
+        field = self.C.build_put_key_data(0x40, [{'key_type': 'tls_psk', 'clear_key': self.PSK_CLEAR,
+                                                  'kcv': self.PSK_KCV}], None)[1:]
+        self.assertEqual(b2h(field), '8511' '10' + b2h(self.PSK_CLEAR) + '03' + b2h(self.PSK_KCV))
+
+
+class PutKey_Length_Test(unittest.TestCase):
+    """Tests for the length of the PUT KEY command APDU.  Lc of GP CardSpec v2.3 Table 11-64 is a
+    single byte, so an oversized key data field cannot be sent."""
+
+    class PutKeyOnly(ADF_SD.AddlShellCommands):
+        """ADF_SD.AddlShellCommands with a canned scc to drive put_key()"""
+        def __init__(self, scp=None, max_cmd_len=255):
+            super().__init__()
+            self.sent = []
+            self.scc = SimpleNamespace(scp=scp, max_cmd_len=max_cmd_len,
+                                       send_apdu_checksw=lambda pdu: (self.sent.append(pdu), ('', '9000'))[1])
+
+        @property
+        def _cmd(self):
+            return SimpleNamespace(lchan=SimpleNamespace(scc=self.scc))
+
+    # KVN, key type, two byte BER length of the key component block, KCV length; KCV suppressed
+    FRAMING = 1 + 1 + 2 + 1
+
+    @staticmethod
+    def key(nbytes: int):
+        return [{'key_type': 'rsa_modulus_n', 'clear_key': bytes(nbytes), 'kcv': b''}]
+
+    def test_lc_matches_data_field(self):
+        # largest key component block that still fits without a secure channel
+        sd = self.PutKeyOnly()
+        sd.put_key(0, 0x40, 1, self.key(255 - self.FRAMING))
+        apdu = sd.sent[0]
+        self.assertEqual(apdu[:8], '80D80001')
+        lc = int(apdu[8:10], 16)
+        self.assertEqual(lc, 255)                       # Lc ...
+        self.assertEqual(len(apdu[10:-2]) // 2, lc)     # ... and it matches the actual data field
+
+    def test_oversized_key_data_raises(self):
+        # real world fat example: RSA-2048 modulus does not fit, led to 3 nibble Lc 106,
+        # which silently shifted and broke the whole APDU by half a byte.
+        sd = self.PutKeyOnly()
+        with self.assertRaises(ValueError) as ctx:
+            sd.put_key(0, 0x40, 1, self.key(256))
+        self.assertIn('262', str(ctx.exception))
+        self.assertIn('255', str(ctx.exception))
+        self.assertEqual(sd.sent, [])                   # nothing was sent to the card
+
+    def test_secure_channel_overhead_lowers_the_limit(self):
+        # scc.max_cmd_len shrinks by the C-MAC + encryption padding of active SCP
+        sd = self.PutKeyOnly(max_cmd_len=239)
+        sd.put_key(0, 0x40, 1, self.key(239 - self.FRAMING))
+        self.assertEqual(int(sd.sent[0][8:10], 16), 239)
+        with self.assertRaises(ValueError):
+            sd.put_key(0, 0x40, 1, self.key(239 - self.FRAMING + 1))
+
+
 class Install_param_Test(unittest.TestCase):
     def test_gen_install_parameters(self):
         load_parameters = gen_install_parameters(256, 256, '010001001505000000000000000000000000')
@@ -298,5 +538,180 @@ class Install_param_Test(unittest.TestCase):
         load_parameters = gen_install_parameters()
         self.assertEqual(load_parameters, 'c900')
 
+class SCP_Overhead_Test(unittest.TestCase):
+    """SCP.overhead varies according to the current security level:
+    C-MAC + at level >= 3 the worst-case padding!
+    """
+
+    def _scp02(self, security_level):
+        scp = SCP02(card_keys=ck_3des_70)
+        scp.sk = Scp02SessionKeys(0x0001, ck_3des_70)
+        scp.security_level = security_level
+        return scp
+
+    def _scp03(self, security_level, s_mode=8):
+        scp = SCP03(card_keys=KEYSET_AES128, s_mode=s_mode)
+        scp.sk = Scp03SessionKeys(KEYSET_AES128, b'\x00' * s_mode, b'\x11' * s_mode)
+        scp.security_level = security_level
+        return scp
+
+    def test_scp02(self):
+        self.assertEqual(self._scp02(0x00).overhead, 0)   # no wrapping at all
+        self.assertEqual(self._scp02(0x01).overhead, 8)   # C-MAC
+        self.assertEqual(self._scp02(0x03).overhead, 16)  # C-MAC + C-DEC: pad80 to 8, largest fit 239
+
+    def test_scp03_s8(self):
+        self.assertEqual(self._scp03(0x00).overhead, 0)
+        self.assertEqual(self._scp03(0x01).overhead, 8)
+        self.assertEqual(self._scp03(0x03).overhead, 16)  # pad80 to 16 within 247 -> 240, minus pad byte
+        self.assertEqual(self._scp03(0x33).overhead, 16)  # R-MAC/R-ENC add no *command* overhead
+
+    def test_scp03_s16(self):
+        self.assertEqual(self._scp03(0x01, s_mode=16).overhead, 16)
+        self.assertEqual(self._scp03(0x03, s_mode=16).overhead, 32)  # pad80 to 16 within 239 -> 224, minus pad byte
+
+
+class SCP_Lc_Limit_Test_Base(unittest.TestCase):
+    """Test wrap_cmd_apdu() boundary handling: data of (255 - overhead) must produce Lc <= 255 else ValueError"""
+
+    def _load_apdu(self, data_len):
+        return h2b('80E80000') + bytes([data_len]) + b'\xa5' * data_len
+
+    def _check_boundary(self, scp):
+        fits = 255 - scp.overhead
+        wrapped = scp.wrap_cmd_apdu(self._load_apdu(fits))
+        self.assertLessEqual(wrapped[4], 255)
+        self.assertEqual(len(wrapped), 5 + wrapped[4])  # case #3: header + Lc bytes, no Le
+        with self.assertRaises(ValueError) as ctx:
+            scp.wrap_cmd_apdu(self._load_apdu(fits + 1))
+        self.assertIn('Lc', str(ctx.exception))
+
+
+class SCP02_Lc_Limit_Test(SCP_Lc_Limit_Test_Base):
+    """Same session vectors as SCP02_Auth_Test"""
+
+    def setUp(self):
+        self.scp02 = SCP02(card_keys=ck_3des_70)
+        self.scp02.gen_init_update_apdu(host_challenge=h2b('40A62C37FA6304F8'))
+        self.scp02.parse_init_update_resp(h2b('00000000000000000000700200016B4524ABEE7CF32EA3838BC148F3'))
+        self.scp02.gen_ext_auth_apdu()
+
+    def test_cmac_only(self):
+        self.scp02.security_level = 0x01
+        self._check_boundary(self.scp02)  # 247 fits, 248 raises
+
+    def test_cmac_cdec(self):
+        self.scp02.security_level = 0x03
+        self._check_boundary(self.scp02)  # 239 fits (-> Lc 248), 240 raises (would be 256)
+
+    def test_cmac_cdec_wrapped_lc(self):
+        # my actual failing case: 240 bytes at level 3
+        self.scp02.security_level = 0x03
+        wrapped = self.scp02.wrap_cmd_apdu(self._load_apdu(239))
+        self.assertEqual(wrapped[4], 248)  # 239 -> pad80 -> 240 ciphertext + 8 mac
+
+
+class SCP03_Lc_Limit_Test(SCP_Lc_Limit_Test_Base):
+    """Session keys derived directly"""
+
+    def _scp03(self, security_level, s_mode):
+        scp = SCP03(card_keys=KEYSET_AES128, s_mode=s_mode)
+        scp.sk = Scp03SessionKeys(KEYSET_AES128, b'\x00' * s_mode, b'\x11' * s_mode)
+        scp.security_level = security_level
+        return scp
+
+    def test_s8_cmac_only(self):
+        self._check_boundary(self._scp03(0x01, 8))    # 247 fits, 248 raises
+
+    def test_s8_cmac_cdec(self):
+        self._check_boundary(self._scp03(0x03, 8))    # 239 fits, 240 raises
+
+    def test_s16_cmac_only(self):
+        self._check_boundary(self._scp03(0x01, 16))   # 239 fits, 240 raises
+
+    def test_s16_cmac_cdec(self):
+        self._check_boundary(self._scp03(0x03, 16))   # 223 fits, 224 raises
+
+
+class _FakeSccForLoad:
+    """mock lchan.scc: records LOAD APDUs, optionally wrapping them through a real SCP
+    instance first where the Lc overflow used to blow up"""
+
+    def __init__(self, max_cmd_len=255, scp=None):
+        self.max_cmd_len = max_cmd_len
+        self.scp = scp
+        self.sent = []
+        self.wrapped = []
+
+    def send_apdu_checksw(self, apdu, sw='9000'):
+        self.sent.append(apdu.lower())
+        if self.scp:
+            self.wrapped.append(self.scp.wrap_cmd_apdu(h2b(apdu)))
+        return ('', '9000')
+
+
+class Load_ChunkLen_Test(unittest.TestCase):
+    """ADF_SD.load() chunking: block size must use scc.max_cmd_len"""
+
+    payload = b'\xaa' * 500  # actual real world case LOAD TLV: C4 + 8201f4 + 500 = 504 total
+
+    def _sd(self, scc):
+        cmd = type('_Cmd', (), {'lchan': type('_Lchan', (), {'scc': scc})(),
+                                'poutput': lambda self, *args: None})()
+        # cmd2 CommandSet has a r/o _cmd property -> shadow it
+        _SD = type('_SD', (ADF_SD.AddlShellCommands,), {'_cmd': cmd})
+        return _SD.__new__(_SD)
+
+    def _blocks(self, scc):
+        """Get (p1, p2, lc) from LOAD APDU"""
+        for apdu in scc.sent:
+            self.assertEqual(apdu[0:4], '80e8')
+            yield int(apdu[4:6], 16), int(apdu[6:8], 16), int(apdu[8:10], 16)
+
+    def test_default_no_scp(self):
+        """Without SCP the old 240 byte block size is kept, no idea what else might rely on this number"""
+        scc = _FakeSccForLoad(max_cmd_len=255)
+        self._sd(scc).load(self.payload)
+        blocks = list(self._blocks(scc))
+        self.assertEqual([b[2] for b in blocks], [240, 240, 24])
+        self.assertEqual([b[0] for b in blocks], [0x00, 0x00, 0x80])  # P1: last block flagged
+        self.assertEqual([b[1] for b in blocks], [0, 1, 2])           # P2: block num
+
+    def test_default_scp02_level3(self):
+        """max_cmd_len 239 (SCP02 lvl 3) squeezes the blocks"""
+        scc = _FakeSccForLoad(max_cmd_len=239)
+        self._sd(scc).load(self.payload)
+        self.assertEqual([b[2] for b in list(self._blocks(scc))], [239, 239, 26])
+
+    def test_explicit_chunk_len(self):
+        scc = _FakeSccForLoad(max_cmd_len=255)
+        self._sd(scc).load(self.payload, chunk_len=100)
+        self.assertEqual([b[2] for b in list(self._blocks(scc))], [100] * 5 + [4])
+
+    def test_explicit_chunk_len_too_large(self):
+        scc = _FakeSccForLoad(max_cmd_len=239)
+        with self.assertRaises(ValueError):
+            self._sd(scc).load(self.payload, chunk_len=240)
+        self.assertEqual(scc.sent, [])  # nothing sent!
+
+    def test_explicit_chunk_len_zero(self):
+        scc = _FakeSccForLoad(max_cmd_len=255)
+        with self.assertRaises(ValueError):
+            self._sd(scc).load(self.payload, chunk_len=0)
+
+    def test_end_to_end_scp02_level3(self):
+        """original failure: 286 byte CAP + SCP02 lvl 3"""
+        scp02 = SCP02(card_keys=ck_3des_70)
+        scp02.gen_init_update_apdu(host_challenge=h2b('40A62C37FA6304F8'))
+        scp02.parse_init_update_resp(h2b('00000000000000000000700200016B4524ABEE7CF32EA3838BC148F3'))
+        scp02.gen_ext_auth_apdu()
+        scp02.security_level = 0x03
+        scc = _FakeSccForLoad(max_cmd_len=255 - scp02.overhead, scp=scp02)
+        self._sd(scc).load(b'\x5a' * 286)
+        self.assertEqual(len(scc.sent), 2)  # 289 byte TLV in blocks of 239
+        for wrapped in scc.wrapped:
+            self.assertLessEqual(wrapped[4], 255)
+
+
 if __name__ == "__main__":
 	unittest.main()

tests/unittests/test_log.py

@@ -37,6 +37,17 @@ expected_message = None
 
 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.