personalization: add GfmSuciRi, GfmSuciCalcInfo, and test

Change-Id: I1b69debf5992aa715171b43b30864dc152dc556f
Jenkins: skip-card-test

pySim/esim/http_json_api.py

@@ -411,14 +411,21 @@ class JsonHttpApiClient():
         # SGP.22, section 6.5.1)
         if response.status_code != self.api_func.expected_http_status:
             raise HttpStatusError(response)
-        if response.content and not response.headers.get('Content-Type').startswith(req_headers['Content-Type']):
+        resp_content_type = response.headers.get('Content-Type')
+        if response.content and not resp_content_type.startswith(req_headers['Content-Type']):
             raise HttpHeaderError(response)
         if not response.headers.get('X-Admin-Protocol', 'gsma/rsp/v2.unknown').startswith('gsma/rsp/v2.'):
             raise HttpHeaderError(response)
 
         # Decode response and return the result back to the caller
         if response.content:
-            output = self.api_func.decode_client(response.json())
+            if resp_content_type.startswith('application/json'):
+                output = self.api_func.decode_client(response.json())
+            elif resp_content_type.startswith('text/plain;charset=UTF-8'):
+                output = { 'data': response.content.decode('utf-8') }
+            else:
+                raise HttpHeaderError(f'unimplemented response Content-Type: {response.headers=!r}')
+
             # In case the response contains a header, check it to make sure that the API call was executed successfully
             # (the presence of the header field is checked by the decode_client method)
             if 'header' in output:

pySim/esim/saip/__init__.py

@@ -1517,8 +1517,11 @@ class ProfileElementHeader(ProfileElement):
     def mandatory_service_add(self, service_name):
         self.decoded['eUICC-Mandatory-services'][service_name] = None
 
+    def mandatory_service_present(self, service_name):
+        return service_name in self.decoded['eUICC-Mandatory-services'].keys()
+
     def mandatory_service_remove(self, service_name):
-        if service_name in self.decoded['eUICC-Mandatory-services'].keys():
+        if self.mandatory_service_present(service_name):
             del self.decoded['eUICC-Mandatory-services'][service_name]
         else:
             raise ValueError("service not in eUICC-Mandatory-services list, cannot remove")
@@ -1774,9 +1777,13 @@ class ProfileElementSequence:
 
         hdr_pe = self.get_pe_for_type('header')
         # patch in the 'manual' services from the existing list:
+        old_svc_set = set()
         for old_svc in hdr_pe.decoded['eUICC-Mandatory-services'].keys():
             if old_svc in manual_services:
-                svc_set.add(old_svc)
+                old_svc_set.add(old_svc)
+        logger.debug(f"{svc_set=}  +  {old_svc_set=}")
+        svc_set = svc_set.union(old_svc_set)
+        logger.debug(f"{svc_set=}")
         hdr_pe.decoded['eUICC-Mandatory-services'] = {x: None for x in svc_set}
 
     def rebuild_mandatory_gfstelist(self):

pySim/esim/saip/batch.py

@@ -20,6 +20,9 @@
 
 import copy
 import pprint
+import logging
+import traceback
+import inspect
 from typing import Generator, Union
 from pySim.esim.saip.personalization import ConfigurableParameter
 from pySim.esim.saip import param_source
@@ -27,6 +30,10 @@ from pySim.esim.saip import ProfileElementSequence, ProfileElementSD
 from pySim.global_platform import KeyUsageQualifier
 from osmocom.utils import b2h
 
+logger = logging.getLogger(__name__)
+def _func_():
+    return inspect.currentframe().f_back.f_code.co_name
+
 # a list of ConfigurableParameter classes and/or ConfigurableParameter class instances
 ParamList = list[Union[type[ConfigurableParameter], ConfigurableParameter]]
 
@@ -121,6 +128,8 @@ class BatchPersonalization:
                     value = p.param_cls.validate_val(input_value)
                     p.param_cls.apply_val(pes, value)
                 except Exception as e:
+                    print(traceback.format_exc())
+                    logger.error('during %s: %r', _func_(), e)
                     raise ValueError(f'{p.param_cls.get_name()} fed by {p.src.name}: {e}') from e
 
             pes.rebuild_mandatory_services()
@@ -324,12 +333,15 @@ class BatchAudit(list):
 
         return batch_audit
 
-    def to_csv_rows(self, headers=True, sort_key=None):
+    def to_csv_rows(self, headers=True, sort_key=None, column_blacklist=None):
         """generator that yields all audits' values as rows, useful feed to a csv.writer."""
         columns = set()
         for audit in self:
             columns.update(audit.keys())
 
+        if column_blacklist:
+            columns.difference_update(set(column_blacklist))
+
         columns = tuple(sorted(columns, key=sort_key))
 
         if headers:

pySim/esim/saip/personalization.py

@@ -19,19 +19,26 @@ import abc
 import enum
 import io
 import re
+import json
 from typing import List, Tuple, Generator, Optional
 
 from construct.core import StreamError
 from osmocom.tlv import camel_to_snake
 from osmocom.utils import hexstr
 from pySim.utils import enc_iccid, dec_iccid, enc_imsi, dec_imsi, h2b, b2h, rpad, sanitize_iccid
-from pySim.ts_31_102 import EF_AD
+from pySim.ts_31_102 import EF_AD, EF_UST, EF_Routing_Indicator, EF_SUCI_Calc_Info, DF_USIM_5GS
 from pySim.ts_51_011 import EF_SMSP
 from pySim.esim.saip import param_source
 from pySim.esim.saip import ProfileElement, ProfileElementSD, ProfileElementSequence
+from pySim.esim.saip import ProfileElementHeader
 from pySim.esim.saip import SecurityDomainKey, SecurityDomainKeyComponent
 from pySim.global_platform import KeyUsageQualifier, KeyType
 
+# optimization: instantiate class instance to get the fid only once.
+file_path_df_5gs = bytes.fromhex(DF_USIM_5GS().fid)
+fid_ri = bytes.fromhex(EF_Routing_Indicator().fid)
+fid_sucici = bytes.fromhex(EF_SUCI_Calc_Info().fid)
+
 def unrpad(s: hexstr, c='f') -> hexstr:
     return hexstr(s.rstrip(c))
 
@@ -290,7 +297,9 @@ class ConfigurableParameter(abc.ABC, metaclass=ClassVarMeta):
            May be overridden by subclasses.
            This default implementation returns the maximum allowed value length -- a good fit for most subclasses.
            '''
-        return cls.get_len_range()[1] or 16
+        l = cls.get_len_range()[1] or 16
+        l = min(10*80, l)
+        return l
 
     @classmethod
     def is_super_of(cls, other_class):
@@ -420,68 +429,70 @@ class BinaryParam(ConfigurableParameter):
 
 
 class EnumParam(ConfigurableParameter):
-    """ConfigurableParameter for named integer enumeration values.
+    """ConfigurableParameter for named value enumerations.
 
-    Subclasses must define a nested enum.IntEnum named 'Values' listing all valid names and their
-    integer codes.  apply_val() and get_values_from_pes() are not implemented here and this must
-    be inherited from another mixin."""
-
-    class Values(enum.IntEnum):
-        pass  # subclasses override this
+    Subclasses define an own value_map, and implement their own apply_val() and get_values_from_pes().
+    """
+    value_map = {
+            # For example:
+            #'Meaningful label for value 23': 0x23,
+            # Where 0x23 is a valid value to use for apply_val(), of any valid type.
+        }
+    _value_map_reverse = None
 
     @classmethod
-    def validate_val(cls, val) -> int:
-        if isinstance(val, int):
-            try:
-                return int(cls.Values(val))
-            except ValueError:
-                pass
-        elif isinstance(val, str):
-            member = cls.map_name_to_val(val, strict=False)
-            if member is not None:
-                return member
+    def validate_val(cls, val):
+        orig_val = val
+        enum_val = None
+        if isinstance(val, str):
+            enum_name = val
+            enum_val = cls.map_name_to_val(enum_name)
 
-        valid = ', '.join(m.name for m in cls.Values)
-        raise ValueError(f"{cls.get_name()}: invalid argument: {val!r}. Valid arguments are: {valid}")
+        # if the str is not one of the known value_map.keys(), is it maybe one of value_map.keys()?
+        if enum_val is None and val in cls.value_map.values():
+            enum_val = val
+
+        if enum_val not in cls.value_map.values():
+            raise ValueError(f"{cls.get_name()}: invalid argument: {orig_val!r}. Valid arguments are:"
+                             f" {', '.join(cls.value_map.keys())}")
+
+        return enum_val
 
     @classmethod
-    def map_name_to_val(cls, name: str, strict=True) -> int:
-        """Return the integer value for a given enum member name. Performs an exact match first,
-        then falls back to fuzzy matching (case-insensitive, punctuation-insensitive)."""
-        try:
-            return int(cls.Values[name])
-        except KeyError:
-            pass
+    def map_name_to_val(cls, name:str, strict=True):
+        val = cls.value_map.get(name)
+        if val is not None:
+            return val
 
-        clean = cls.clean_name_str(name)
-        for member in cls.Values:
-            if cls.clean_name_str(member.name) == clean:
-                return int(member)
+        clean_name = cls.clean_name_str(name)
+        for k, v in cls.value_map.items():
+            if clean_name == cls.clean_name_str(k):
+                return v
 
         if strict:
-            valid = ', '.join(m.name for m in cls.Values)
-            raise ValueError(f"{cls.get_name()}: {name!r} is not a known value. Known values are: {valid}")
+            raise ValueError(f"Problem in {cls.get_name()}: {name!r} is not a known value."
+                    f" Known values are: {cls.value_map.keys()!r}")
         return None
 
     @classmethod
     def map_val_to_name(cls, val, strict=False) -> str:
-        """Return the enum member name for a given integer value."""
-        try:
-            return cls.Values(val).name
-        except ValueError:
-            if strict:
-                raise ValueError(f"{cls.get_name()}: {val!r} ({type(val).__name__}) is not a known value.")
-            return None
+        if cls._value_map_reverse is None:
+            cls._value_map_reverse = dict((v, k) for k, v in cls.value_map.items())
+
+        name = cls._value_map_reverse.get(val)
+        if name:
+            return name
+        if strict:
+            raise ValueError(f"Problem in {cls.get_name()}: {val!r} ({type(val)}) is not a known value."
+                    f" Known values are: {cls.value_map.values()!r}")
+        return None
 
     @classmethod
-    def name_normalize(cls, name: str) -> str:
-        """Map a (possibly fuzzy) name to its canonical enum member name."""
-        return cls.Values(cls.map_name_to_val(name)).name
+    def name_normalize(cls, name:str) -> str:
+        return cls.map_val_to_name(cls.map_name_to_val(name))
 
     @classmethod
-    def clean_name_str(cls, val: str) -> str:
-        """Strip punctuation and case for fuzzy name comparison.
-        Treats hyphens and underscores as equivalent (both removed)."""
+    def clean_name_str(cls, val):
         return re.sub('[^0-9A-Za-z]', '', val).lower()
 
 
@@ -635,13 +646,14 @@ class SmspTpScAddr(ConfigurableParameter):
             # - To generate the right amount of fillFileContent, pass total_len=42 to encode_record_bin().
             # - To show the right size in the PES, set f_smsp.rec_len = 42
             ef_smsp_dec['alpha_id'] = ''
-            f_smsp.rec_len = 42
+
+            # we can set this to choose a fixed length:
+            #f_smsp.rec_len = 42
+            # but leave rec_len unchanged to keep the same length as was found in the eSIM template.
 
             # re-encode into the File body.
-            #
-            #print("SMSP  (new): %s" % f_smsp.body)
-            # re-generate the pe.decoded member from the File instance
             f_smsp.body = ef_smsp.encode_record_bin(ef_smsp_dec, 1, total_len=f_smsp.rec_len)
+            # re-generate the pe.decoded member from the File instance
             pe.file2pe(f_smsp)
 
     @classmethod
@@ -663,69 +675,57 @@ class SmspTpScAddr(ConfigurableParameter):
 
 
 class MncLen(EnumParam):
-    """MNC length.  Sets only the MNC length field in EF.AD (Administrative Data).
-    Accepted values: integer 2 or 3, digit strings '2' or '3', or enum names 'MNC2'/'MNC3'.
-    """
+    """MNC length. Must be either 2 or 3. Sets only the MNC length field in EF-AD (Administrative Data)."""
     name = 'MNC-LEN'
-    example_input = '2'
+    value_map = { '2': 2, '3': 3 }
     default_source = param_source.ConstantSource
-
-    class Values(enum.IntEnum):
-        MNC2 = 2
-        MNC3 = 3
+    example_input = '2'
 
     @classmethod
-    def validate_val(cls, val):
-        if isinstance(val, str) and val.isdigit():
-            val = int(val)
-        return super().validate_val(val)
-
-    @classmethod
-    def _get_f_ad(cls, pe: ProfileElement):
-        if not hasattr(pe, 'files'):
-            return None
-        f_ad = pe.files.get('ef-ad', None)
-        if f_ad and f_ad.body:
-            return f_ad
-        return None
-
-    @classmethod
-    def _decode_f_ad(cls, f_ad):
-        try:
-            ef_ad_dec = EF_AD().decode_bin(f_ad.body)
-        except StreamError:
-            return None
-        if 'mnc_len' not in ef_ad_dec:
-            return None
-        return ef_ad_dec
-
-    @classmethod
-    def apply_val(cls, pes: ProfileElementSequence, val: int):
+    def apply_val(cls, pes: ProfileElementSequence, val):
+        """val must be an int: either 2 or 3"""
         for pe in pes.get_pes_for_type('usim'):
-            f_ad = cls._get_f_ad(pe)
-            if f_ad is None:
+            if not hasattr(pe, 'files'):
+                continue
+            f_ad = pe.files.get('ef-ad')
+            if not f_ad:
                 continue
             # decode existing values
-            ef_ad_dec = cls._decode_f_ad(f_ad)
-            if ef_ad_dec is None:
+            if not f_ad.body:
+                continue
+            try:
+                ef_ad = EF_AD()
+                ef_ad_dec = ef_ad.decode_bin(f_ad.body)
+            except StreamError:
+                continue
+            if 'mnc_len' not in ef_ad_dec:
                 continue
             # change mnc_len
             ef_ad_dec['mnc_len'] = val
             # re-encode into the File body
-            f_ad.body = EF_AD().encode_bin(ef_ad_dec)
+            f_ad.body = ef_ad.encode_bin(ef_ad_dec)
             pe.file2pe(f_ad)
 
     @classmethod
     def get_values_from_pes(cls, pes: ProfileElementSequence):
         for pe in pes.get_pes_for_type('usim'):
-            f_ad = cls._get_f_ad(pe)
+            if not hasattr(pe, 'files'):
+                continue
+            f_ad = pe.files.get('ef-ad', None)
             if f_ad is None:
                 continue
-            ef_ad_dec = cls._decode_f_ad(f_ad)
-            if ef_ad_dec is None:
+
+            try:
+                ef_ad = EF_AD()
+                ef_ad_dec = ef_ad.decode_bin(f_ad.body)
+            except StreamError:
                 continue
-            mnc_len = ef_ad_dec.get('mnc_len')
-            yield { cls.name: str(mnc_len) }
+
+            mnc_len = ef_ad_dec.get('mnc_len', None)
+            if mnc_len is None:
+                continue
+
+            yield { cls.name: cls.map_val_to_name(int(mnc_len)) }
 
 
 class SdKey(BinaryParam):
@@ -1099,17 +1099,17 @@ class AlgorithmID(EnumParam, AlgoConfig):
     """use validate_val() from EnumParam, and apply_val() from AlgoConfig.
     In get_values_from_pes(), return enum value names, not raw values."""
     name = "Algorithm"
+    # as in pySim/esim/asn1/saip/PE_Definitions-3.3.1.asn
+    value_map = {
+            "Milenage" : 1,
+            "TUAK" : 2,
+            "usim-test" : 3,
+        }
     algo_config_key = 'algorithmID'
     example_input = "Milenage"
     default_source = param_source.ConstantSource
 
-    # as in pySim/esim/asn1/saip/PE_Definitions-3.3.1.asn
-    class Values(enum.IntEnum):
-        Milenage  = 1
-        TUAK      = 2
-        usim_test = 3  # input 'usim-test' also accepted via fuzzy matching
-
-    # EnumParam.validate_val() returns the int values from Values
+    # EnumParam.validate_val() returns the int values from value_map
 
     @classmethod
     def get_values_from_pes(cls, pes: ProfileElementSequence):
@@ -1184,3 +1184,411 @@ class TuakNumberOfKeccak(IntegerParam, AlgoConfig):
     max_val = 255
     example_input = '1'
     default_source = param_source.ConstantSource
+    numeric_base = None  # indicate that this won't need random number sources
+
+
+class EfUstServiceParam(EnumParam):
+    """superclass for EF-UST service flag parameters"""
+    service_idx = 0
+    value_map = { 'enabled': True, 'disabled': False }
+    default_source = param_source.ConstantSource
+    example_input = sorted(value_map.keys())[0]
+
+    @classmethod
+    def apply_val(cls, pes: ProfileElementSequence, val):
+        for pe in pes.get_pes_for_type('usim'):
+            f_ust = pe.files['ef-ust']
+            ef_ust = EF_UST()
+            ust = ef_ust.decode_bin(f_ust.body)
+
+            ust[cls.service_idx]['activated'] = val
+
+            f_ust.body = ef_ust.encode_bin(ust)
+            pe.file2pe(f_ust)
+
+    @classmethod
+    def get_values_from_pes(cls, pes: ProfileElementSequence):
+        for pe in pes.get_pes_for_type('usim'):
+            f_ust = pe.files.get('ef-ust', None)
+            if not f_ust:
+                continue
+            ef_ust = EF_UST()
+            try:
+                ust = ef_ust.decode_bin(f_ust.body)
+
+                service_flag = ust[cls.service_idx]['activated']
+                yield { cls.name: cls.map_val_to_name(service_flag) }
+            except:
+                pass
+
+class SuciActive(EfUstServiceParam):
+    """EF-UST service nr 124: enable or disable the SUCI service."""
+    service_idx = 124
+    name = '5G-SUCI-active'
+    value_map = { 'SUCI-off': False, 'SUCI-on': True }
+    example_input = 'SUCI-on'
+
+class SuciInUsim(EfUstServiceParam):
+    """EF-UST service nr 125: calculate SUCI in UE or in USIM"""
+    service_idx = 125
+    name = '5G-SUCI-in-USIM'
+    value_map = { 'SUCI-in-UE': False, 'SUCI-in-USIM': True }
+    example_input = 'SUCI-in-USIM'
+
+class SuciRi(ConfigurableParameter):
+    """SUCI Routing Indicator as in section 4.4.11.11 of 3GPP TS 31.102"""
+    name = '5G-SUCI-RI'
+    allow_chars = '0123456789'
+    min_len = 1
+    max_len = 4
+    allow_types = (str,)
+    example_input = '0'
+    default_source = param_source.ConstantSource
+
+    KEY_RI = "routing_indicator"
+
+    @classmethod
+    def apply_val(cls, pes: ProfileElementSequence, val):
+        for pe in pes.get_pes_for_type('df-5gs'):
+            f_ri = pe.files.get('ef-routing-indicator', None)
+            if f_ri is None:
+                continue
+            ef_ri = EF_Routing_Indicator()
+            ri = ef_ri.decode_bin(f_ri.body)
+
+            ri[cls.KEY_RI] = str(val)
+
+            f_ri.body = ef_ri.encode_bin(ri)
+            pe.file2pe(f_ri)
+
+    @classmethod
+    def get_values_from_pes(cls, pes: ProfileElementSequence):
+        for pe in pes.get_pes_for_type('df-5gs'):
+            f_ri = pe.files.get('ef-routing-indicator', None)
+            if f_ri is None:
+                continue
+            ef_ri = EF_Routing_Indicator()
+            try:
+                ri = ef_ri.decode_bin(f_ri.body)
+                yield { cls.name: ri.get(cls.KEY_RI) }
+            except:
+                pass
+
+class SuciCalcInfoParameter(ConfigurableParameter):
+    """SUCI Calculation Information as in section 4.4.11.8 of 3GPP TS 31.102"""
+    name = '5G-SUCI-CalcInfo'
+    default_source = param_source.ConstantSource
+    allow_types = (str,)
+    max_len = 4096  # to indicate a large input field to UI renderers
+    example_input = '{"prot_scheme_id_list": [], "hnet_pubkey_list": []}'
+
+    PE_IN_UE = ("df-5gs", "ef-suci-calc-info")
+    PE_IN_USIM = ("df-saip", "ef-suci-calc-info-usim")
+    suci_calc_info_pe = None
+
+    @classmethod
+    def validate_val(cls, val):
+        val = super().validate_val(val)
+
+        if not val:
+            val = "{}"
+
+        # check that it is a dict something like
+        # {
+        #     "prot_scheme_id_list": [
+        #         {"priority": 0, "identifier": 2, "key_index": 1},
+        #         {"priority": 1, "identifier": 1, "key_index": 2},
+        #         ],
+        #     "hnet_pubkey_list": [
+        #         {"hnet_pubkey_identifier": 27,
+        #          "hnet_pubkey": "0472DA71976234CE833A6907425867B82E074D44EF907DFB4B3E21C1C2256EBCD15A7DED52FCBB097A4ED250E036C7B9C8C7004C4EEDC4F068CD7BF8D3F900E3B4"},
+        #         {"hnet_pubkey_identifier": 30,
+        #          "hnet_pubkey": "5A8D38864820197C3394B92613B20B91633CBD897119273BF8E4A6F4EEC0A650"},
+        #          ],
+        # }
+
+        try:
+            d = json.loads(val)
+        except json.decoder.JSONDecodeError as e:
+            raise ValueError(f"Cannot parse SUCI Calc Info: {e}") from e
+
+        KEY_PSI_LIST = 'prot_scheme_id_list'
+        KEY_HPK_LIST = 'hnet_pubkey_list'
+        KEYS_D = set((KEY_HPK_LIST, KEY_PSI_LIST))
+        KEYS_PSI = set(('identifier', 'key_index', 'priority'))
+        KEYS_HPK = set(('hnet_pubkey_identifier', 'hnet_pubkey'))
+
+        if not d:
+            d = { KEY_PSI_LIST: [], KEY_HPK_LIST: [] }
+
+        if not (isinstance(d, dict)
+                and set(d.keys()) == KEYS_D):
+            raise ValueError(f"Unexpected structure in SUCI Calc Info: expected dict with entries {KEYS_D}")
+
+        psi = d.get(KEY_PSI_LIST, None)
+        if not all((set(e.keys()) == KEYS_PSI) for e in psi):
+            raise ValueError("Unexpected structure in SUCI Calc Info:"
+                             f" in {KEY_PSI_LIST}, expected dict with entries {KEYS_PSI}")
+
+        hpk = d.get(KEY_HPK_LIST, None)
+        if not all((set(e.keys()) == KEYS_HPK) for e in hpk):
+            raise ValueError("Unexpected structure in SUCI Calc Info:"
+                             f" in {KEY_HPK_LIST}, expected dict with entries {KEYS_HPK}")
+        return d
+
+    @classmethod
+    def _apply_suci(cls, pes: ProfileElementSequence, val, pe_type="df-5gs", pe_file="ef-suci-calc-info"):
+        for pe in pes.get_pes_for_type(pe_type):
+            f_sucici = pe.files.get(pe_file, None)
+            if not f_sucici:
+                continue
+            ef_sucici = EF_SUCI_Calc_Info()
+            body = ef_sucici.encode_bin(val)
+
+            # 0xff pad up to the existing file size, so that the underlying template doesn't come through
+            is_size = f_sucici.file_size
+            pad_n = is_size - len(body)
+            if pad_n > 0:
+                body = body + b'\xff' * pad_n
+
+            f_sucici.body = body
+            pe.file2pe(f_sucici)
+
+    @classmethod
+    def apply_val(cls, pes: ProfileElementSequence, val):
+        cls._apply_suci(pes, val, *cls.suci_calc_info_pe)
+
+    @staticmethod
+    def normalize_sucici(sucici:dict):
+        """Normalize the CalcInfo dict so it can be json encoded:
+           convert bytes to hex strings."""
+        if not sucici:
+            sucici = {}
+
+        for hnet_pubkey in sucici.get('hnet_pubkey_list', ()):
+            val = hnet_pubkey['hnet_pubkey']
+            if isinstance(val, bytes):
+                val = b2h(val)
+                hnet_pubkey['hnet_pubkey'] = val
+
+        return sucici
+
+    @classmethod
+    def _get_suci(cls, pes: ProfileElementSequence, pe_type="df-5gs", pe_file="ef-suci-calc-info"):
+        for pe in pes.get_pes_for_type(pe_type):
+            f_sucici = pe.files.get(pe_file, None)
+            if not f_sucici:
+                continue
+            ef_sucici = EF_SUCI_Calc_Info()
+            sucici = ef_sucici.decode_bin(f_sucici.body)
+
+            # normalize to string (bytes cannot go into json)
+            sucici = cls.normalize_sucici(sucici)
+
+            yield { cls.name: json.dumps(sucici) }
+
+    @classmethod
+    def get_values_from_pes(cls, pes: ProfileElementSequence):
+        yield from cls._get_suci(pes, *cls.suci_calc_info_pe)
+
+class SuciCalcInfoUe(SuciCalcInfoParameter):
+    """SUCI Calculation Information as in section 4.4.11.8 of 3GPP TS 31.102, readable by UE (DF-5GS)"""
+    name = '5G-SUCI-CalcInfo-UE'
+    suci_calc_info_pe = SuciCalcInfoParameter.PE_IN_UE
+
+class SuciCalcInfoUsim(SuciCalcInfoParameter):
+    """SUCI Calculation Information as in section 4.4.11.8 of 3GPP TS 31.102, readable only by USIM (DF-SAIP)"""
+    name = '5G-SUCI-CalcInfo-USIM'
+    suci_calc_info_pe = SuciCalcInfoParameter.PE_IN_USIM
+
+def gfm_find(pes: ProfileElementSequence, file_path:bytes, ef_fid:bytes):
+    """look through genericFileManagement PE and return the fmc list with start and end indexes as
+       (fmc_list, first_idx, after_last_idx)
+       so that fmc_list[first_idx:after_last_idx] is the slice of file management commands relevant to the given
+       file_path/ef_fid.
+    """
+    for pe in pes.get_pes_for_type('genericFileManagement'):
+        path_match = False
+        creating_fid = False
+
+        for fmc in pe.decoded['fileManagementCMD']:
+            first = None
+            last = None
+
+            for idx in range(len(fmc)):
+                cmd, arg = fmc[idx]
+
+                if cmd == 'filePath':
+                    path_match = (arg == file_path)
+                    if not path_match:
+                        creating_fid = False
+                elif path_match and cmd == 'createFCP':
+                    creating_fid = (arg.get('fileID') == ef_fid)
+                if creating_fid:
+                    if first is None:
+                        first = idx
+                        last = idx
+                    first = min(first, idx)
+                    last = max(last, idx)
+
+            if first is not None:
+                yield fmc, first, last + 1
+
+# genericFileManagement 5G params
+
+def pes_get_adf_fid(pes:ProfileElementSequence, naa_name="usim", adf_name="adf-usim"):
+    adf = pes.get_pe_for_type(naa_name)
+    return adf.decoded[adf_name][0][1]['fileID']
+
+def mk_adf_df_path(pes, naa:str, adf:str, file_path:bytes) -> bytes:
+    adf_file_id = pes_get_adf_fid(pes, naa, adf)
+    return b''.join((adf_file_id, file_path))
+
+def gfm_get_file_content(pes: ProfileElementSequence, naa:str, adf:str, file_path:bytes, ef_fid:bytes) -> bytes:
+    '''find a given file in the genericFileManagement section, and return the bytes from the first fillFileContent
+       item.
+       TODO: implement File.from_gfm() and return the full resulting bytes?
+    '''
+    adf_df_path = mk_adf_df_path(pes, naa, adf, file_path)
+
+    data = []
+    for fmc, first_idx, after_last_idx in gfm_find(pes, adf_df_path, ef_fid):
+        assert fmc[first_idx][0] == 'createFCP'
+        assert after_last_idx > first_idx
+
+        idx = first_idx + 1
+        while idx < after_last_idx:
+            if fmc[idx][0] == 'fillFileContent':
+                data.append(fmc[idx][1])
+            idx += 1
+
+    return data
+
+def gfm_set_file_content(pes: ProfileElementSequence, naa:str, adf:str, file_path:bytes, ef_fid:bytes, file_content:bytes) -> int:
+    adf_df_path = mk_adf_df_path(pes, naa, adf, file_path)
+
+    found = 0
+    for fmc, first_idx, after_last_idx in gfm_find(pes, adf_df_path, ef_fid):
+        assert fmc[first_idx][0] == 'createFCP'
+        assert after_last_idx > first_idx
+
+        new_fmc = [
+                fmc[first_idx],
+                ('fillFileContent', file_content),
+            ]
+        new_fmc[0][1]['efFileSize'] = bytes((len(file_content), ))
+
+        fmc[first_idx:after_last_idx] = new_fmc
+
+        found += 1
+    return found
+
+class GfmSuciRi(SuciRi):
+    """SUCI Routing Indicator as in section 4.4.11.11 of 3GPP TS 31.102,
+       applied via General File Management. Intended for SAIP 2.1 profiles."""
+    name = 'GFM-5G-SUCI-RI'
+
+    @classmethod
+    def apply_val(cls, pes: ProfileElementSequence, val):
+        ri = {
+            "routing_indicator": str(val),
+            "rfu": "ffff"
+        }
+        ef_ri = EF_Routing_Indicator()
+        found = gfm_set_file_content(pes, 'usim', 'adf-usim', file_path_df_5gs, fid_ri,
+                                     ef_ri.encode_bin(ri))
+        if not found:
+            raise ValueError(f"No target file found, Cannot apply {cls.name} = {ri}")
+
+        data = gfm_get_file_content(pes, 'usim', 'adf-usim', file_path_df_5gs, fid_ri)
+        val = ef_ri.decode_bin(b''.join(data))
+
+    @classmethod
+    def get_values_from_pes(cls, pes: ProfileElementSequence):
+        data = gfm_get_file_content(pes, 'usim', 'adf-usim', file_path_df_5gs, fid_ri)
+        if not data:
+            return
+
+        data = b''.join(data)
+        if not data:
+            return
+
+        ef_ri = EF_Routing_Indicator()
+        ri = ef_ri.decode_bin(data)
+        yield { cls.name: ri.get(cls.KEY_RI) }
+
+class GfmSuciCalcInfoUe(SuciCalcInfoUe):
+    """SUCI Calculation Information as in section 4.4.11.8 of 3GPP TS 31.102, readable by UE (DF-5GS),
+       applied via General File Management. Intended for SAIP 2.1 profiles."""
+    name = 'GFM-5G-SUCI-CalcInfo-UE'
+
+    @classmethod
+    def apply_val(cls, pes: ProfileElementSequence, val):
+        if not isinstance(val, dict):
+            raise ValueError("val should be a dict, after 'val = SuciCalcInfoParameter.validate_val(val)'")
+
+        ef_sucici = EF_SUCI_Calc_Info()
+        body = ef_sucici.encode_bin(val)
+        gfm_set_file_content(pes, 'usim', 'adf-usim', file_path_df_5gs, fid_sucici,
+                             body)
+
+    @classmethod
+    def get_values_from_pes(cls, pes: ProfileElementSequence):
+        data = gfm_get_file_content(pes, 'usim', 'adf-usim', file_path_df_5gs, fid_sucici)
+        if not data:
+            return
+
+        data = b''.join(data)
+        if not data:
+            return
+
+        ef_sucici = EF_SUCI_Calc_Info()
+        sucici = ef_sucici.decode_bin(data)
+        sucici = cls.normalize_sucici(sucici)
+        yield { cls.name: json.dumps(sucici) }
+
+
+class EuiccMandatoryServiceParam(EnumParam):
+    """superclass for managing items of the ProfileHeader / eUICC-Mandatory-services ServicesList"""
+    service_name = None
+    value_map = { 'mandatory': True, 'optional': False }
+    default_source = param_source.ConstantSource
+    example_input = sorted(value_map.keys())[0]
+
+    @classmethod
+    def apply_val(cls, pes: ProfileElementSequence, val):
+        for pe in pes.get_pes_for_type('header'):
+            assert isinstance(pe, ProfileElementHeader)
+            if val:
+                pe.mandatory_service_add(cls.service_name)
+            else:
+                # explicitly check to avoid exception when then service is already not present
+                if pe.mandatory_service_present(cls.service_name):
+                    pe.mandatory_service_remove(cls.service_name)
+
+    @classmethod
+    def get_values_from_pes(cls, pes: ProfileElementSequence):
+        for pe in pes.get_pes_for_type('header'):
+            assert isinstance(pe, ProfileElementHeader)
+            val = bool(pe.mandatory_service_present(cls.service_name))
+            yield { cls.name: cls.map_val_to_name(val) }
+
+class EuiccMandatoryServiceGetIdentity(EuiccMandatoryServiceParam):
+    """eUICC Mandatory Services: get-identity. The eUICC must be capable of providing a 5G identity using SUCI-CalcInfo
+    located in the USIM's DF-SAIP, see parameter 5G-SUCI-CalcInfo-USIM."""
+    name = '5G-eUICC-get-identity'
+    service_name = 'get-identity'
+
+class EuiccMandatoryServiceProfileA(EuiccMandatoryServiceParam):
+    """eUICC Mandatory Services: profile-a-x25519. The eUICC must be able to estblish a 5G identity using an X25519 key,
+    as provided in a profile-A ("identifier": 1) key in SUCI-CalcInfo located in the USIM's DF-SAIP, see parameter
+    5G-SUCI-CalcInfo-USIM."""
+    name = '5G-eUICC-profile-a-x25519'
+    service_name = 'profile-a-x25519'
+
+class EuiccMandatoryServiceProfileB(EuiccMandatoryServiceParam):
+    """eUICC Mandatory Services: profile-b-p256. The eUICC must be able to estblish a 5G identity using a P256 key, as
+    provided in a profile-B ("identifier": 2) key in SUCI-CalcInfo located in the USIM's DF-SAIP, see parameter
+    5G-SUCI-CalcInfo-USIM."""
+    name = '5G-eUICC-profile-b-p256'
+    service_name = 'profile-b-p256'

pySim/global_platform/__init__.py

@@ -18,12 +18,10 @@ 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 *
@@ -150,24 +148,6 @@ 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
@@ -622,8 +602,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, AES and TLS-PSK keys.  You can suppress the
-            latter using `--suppress-key-check`.
+            otherwise be automatically generated for DES and AES 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
@@ -640,81 +620,33 @@ 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 = h2b(opts.key_check[i])
+                    kcv = opts.key_check[i]
                 elif opts.suppress_key_check:
-                    kcv = b''
+                    kcv = ''
                 else:
-                    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})
+                    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})
             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).  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))
+        # Table 11-68: Key Data Field - Format 1 (Basic Format)
+        KeyDataBasic = GreedyRange(Struct('key_type'/KeyType,
+                                          'kcb'/Prefixed(Int8ub, GreedyBytes),
+                                          'kcv'/Prefixed(Int8ub, GreedyBytes)))
 
-        @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:
+        def put_key(self, old_kvn:int, kvn: int, kid: int, key_dict: 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 = 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))
+            key_data = kvn.to_bytes(1, 'big') + build_construct(ADF_SD.AddlShellCommands.KeyDataBasic, key_dict)
             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
@@ -894,32 +826,23 @@ 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), opts.chunk_len)
+                self.load(h2b(opts.from_hex))
             elif opts.from_file is not None:
-                self.load(opts.from_file.read(), opts.chunk_len)
+                self.load(opts.from_file.read())
             elif opts.from_cap_file is not None:
                 cap = CapFile(opts.from_cap_file)
-                self.load(cap.get_loadfile(), opts.chunk_len)
+                self.load(cap.get_loadfile())
             else:
                 raise ValueError('load source not specified!')
 
-        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)
+        def load(self, contents:bytes, chunk_len:int = 240):
+            # TODO:tune chunk_len based on the overhead of the used SCP?
             # 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
@@ -958,8 +881,6 @@ 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):
@@ -998,7 +919,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, opts.chunk_len)
+            self.load(load_file)
             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))
@@ -1144,16 +1065,10 @@ 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,29 +182,6 @@ 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)
 
@@ -238,20 +215,11 @@ 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."""
-        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))
+        num_pad = len(key) % self.sk.blocksize
+        if num_pad:
+            return bertlv_encode_len(len(key)) + self.dek_encrypt(key + b'\x00'*num_pad)
         return self.dek_encrypt(key)
 
     def decrypt_key(self, encrypted_key:bytes) -> bytes:
@@ -264,8 +232,9 @@ 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.
-            key_len, remainder = bertlv_parse_len(encrypted_key)
-            return self.dek_decrypt(remainder)[:key_len]
+            decrypted = self.dek_decrypt(encrypted_key)
+            key_len, remainder = bertlv_parse_len(decrypted)
+            return 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
@@ -291,8 +260,10 @@ 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]]
-    # C-MAC (Single DES + final 3DES, B.1.2.2) is always one full DES block
-    mac_len = 8
+
+    def __init__(self, *args, **kwargs):
+        self.overhead = 8
+        super().__init__(*args, **kwargs)
 
     def dek_encrypt(self, plaintext:bytes) -> bytes:
         # See also GPC section B.1.1.2, E.4.7, and E.4.1
@@ -367,16 +338,10 @@ 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(padded_data)
+            data = k.encrypt(pad80(data, 8))
             lc = len(data)
 
         lc += 8
@@ -512,13 +477,9 @@ 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,7 +17,6 @@ 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
@@ -336,8 +335,6 @@ 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)
 
@@ -384,7 +381,7 @@ class _AM_DO_DF(DataObject):
 
 
 class _AM_DO_EF(DataObject):
-    """ISO7816-4:2005 5.4.3.1 Table 17"""
+    """ISO7816-4 9.3.2 Table 18 + 9.3.3.1 Table 31"""
 
     def __init__(self):
         super().__init__('access_mode', 'Access Mode', tag=0x80)
@@ -432,7 +429,7 @@ class _AM_DO_EF(DataObject):
 
 
 class _AM_DO_CHDR(DataObject):
-    """Command Header Access Mode DO according to ISO 7816-4:2005 5.4.3.2 Table 22."""
+    """Command Header Access Mode DO according to ISO 7816-4 Table 32."""
 
     def __init__(self, tag):
         super().__init__('command_header', 'Command Header Description', tag=tag)
@@ -546,9 +543,8 @@ 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)
 
@@ -736,57 +732,36 @@ 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, but our caller may
-        # be able to pass us a hint:
-        arr_seq = self.__get_do_sequence(kwargs.get('decode_for_df', False))
+        # we can only guess if we should decode for EF or DF here :(
+        arr_seq = DataObjectSequence('arr', sequence=[AM_DO_EF, SC_DO])
         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, but our caller may
-        # be able to pass us a hint:
-        arr_seq = self.__get_do_sequence(kwargs.get('encode_for_df', False))
+        # we can only guess if we should decode for EF or DF here :(
+        arr_seq = DataObjectSequence('arr', sequence=[AM_DO_EF, SC_DO])
         return arr_seq.encode_multi(in_json)
 
     @with_default_category('File-Specific Commands')
     class AddlShellCommands(CommandSet):
-        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)
+        @cmd2.with_argparser(LinFixedEF.ShellCommands.read_rec_dec_parser)
         def do_read_arr_record(self, opts):
             """Read one EF.ARR record in flattened, human-friendly form."""
-            (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, _sw) = self._cmd.lchan.read_record_dec(opts.RECORD_NR)
             data = self._cmd.lchan.selected_file.flatten(data)
             self._cmd.poutput_json(data, opts.oneline)
 
-        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)
+        @cmd2.with_argparser(LinFixedEF.ShellCommands.read_recs_dec_parser)
         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):
-                (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, _sw) = self._cmd.lchan.read_record_dec(recnr)
                 data = self._cmd.lchan.selected_file.flatten(data)
                 data_list.append(data)
             self._cmd.poutput_json(data_list, opts.oneline)

tests/unittests/test_configurable_parameters.py

@@ -21,6 +21,7 @@ import enum
 import io
 import sys
 import unittest
+import json
 from importlib import resources
 from osmocom.utils import hexstr
 from pySim.esim.saip import ProfileElementSequence
@@ -53,16 +54,21 @@ class ConfigurableParameterTest(unittest.TestCase):
     def test_parameters(self):
 
         upp_fnames = (
+                'SAIP2.1_gfmsuci.der',
                 'TS48v5_SAIP2.1A_NoBERTLV.der',
                 'TS48v5_SAIP2.3_BERTLV_SUCI.der',
                )
 
         class Paramtest:
-            def __init__(self, param_cls, val, expect_val, expect_clean_val=None):
+            iff_present_default = False
+            def __init__(self, param_cls, val, expect_val, expect_clean_val=None, iff_present=None):
                 self.param_cls = param_cls
                 self.val = val
                 self.expect_clean_val = expect_clean_val
                 self.expect_val = expect_val
+                if iff_present is None:
+                    iff_present = Paramtest.iff_present_default
+                self.iff_present = iff_present
 
         param_tests = [
             Paramtest(param_cls=p13n.Imsi, val='123456',
@@ -148,7 +154,7 @@ class ConfigurableParameterTest(unittest.TestCase):
             Paramtest(param_cls=p13n.AlgorithmID,
                       val='usim-test',
                       expect_clean_val=3,
-                      expect_val='usim_test'),
+                      expect_val='usim-test'),
 
             Paramtest(param_cls=p13n.AlgorithmID,
                       val=1,
@@ -161,7 +167,7 @@ class ConfigurableParameterTest(unittest.TestCase):
             Paramtest(param_cls=p13n.AlgorithmID,
                       val=3,
                       expect_clean_val=3,
-                      expect_val='usim_test'),
+                      expect_val='usim-test'),
 
             Paramtest(param_cls=p13n.K,
                       val='01020304050607080910111213141516',
@@ -274,7 +280,102 @@ class ConfigurableParameterTest(unittest.TestCase):
                       expect_clean_val=3,
                       expect_val='3'),
 
-            ]
+            Paramtest(param_cls=p13n.EuiccMandatoryServiceGetIdentity,
+                      val='mandatory',
+                      expect_clean_val=True,
+                      expect_val='mandatory'),
+            Paramtest(param_cls=p13n.EuiccMandatoryServiceGetIdentity,
+                      val='optional',
+                      expect_clean_val=False,
+                      expect_val='optional'),
+
+            Paramtest(param_cls=p13n.EuiccMandatoryServiceProfileA,
+                      val='mandatory',
+                      expect_clean_val=True,
+                      expect_val='mandatory'),
+            Paramtest(param_cls=p13n.EuiccMandatoryServiceProfileA,
+                      val='optional',
+                      expect_clean_val=False,
+                      expect_val='optional'),
+
+            Paramtest(param_cls=p13n.EuiccMandatoryServiceProfileB,
+                      val='mandatory',
+                      expect_clean_val=True,
+                      expect_val='mandatory'),
+            Paramtest(param_cls=p13n.EuiccMandatoryServiceProfileB,
+                      val='optional',
+                      expect_clean_val=False,
+                      expect_val='optional'),
+        ]
+
+        Paramtest.iff_present_default = True
+
+        sucici = {
+             "prot_scheme_id_list": [
+                 {"priority": 0, "identifier": 2, "key_index": 1},
+                 {"priority": 1, "identifier": 1, "key_index": 2},
+                 ],
+             "hnet_pubkey_list": [
+                 {"hnet_pubkey_identifier": 27,
+                  "hnet_pubkey": "0472da71976234ce833a6907425867b82e074d44ef907dfb4b3e21c1c2256ebcd15a7ded52fcbb097a4ed250e036c7b9c8c7004c4eedc4f068cd7bf8d3f900e3b4"},
+                 {"hnet_pubkey_identifier": 30,
+                  "hnet_pubkey": "5a8d38864820197c3394b92613b20b91633cbd897119273bf8e4a6f4eec0a650"},
+                  ],
+        }
+
+        param_tests.extend([
+            Paramtest(param_cls=p13n.SuciActive, val='SUCI-on',
+                      expect_clean_val=True,
+                      expect_val={'5G-SUCI-active': 'SUCI-on'}),
+            Paramtest(param_cls=p13n.SuciActive, val='SUCI-off',
+                      expect_clean_val=False,
+                      expect_val={'5G-SUCI-active': 'SUCI-off'}),
+
+            Paramtest(param_cls=p13n.SuciInUsim, val='SUCI-in-UE',
+                      expect_clean_val=False,
+                      expect_val={'5G-SUCI-in-USIM': 'SUCI-in-UE'}),
+            Paramtest(param_cls=p13n.SuciInUsim, val='SUCI-in-USIM',
+                      expect_clean_val=True,
+                      expect_val={'5G-SUCI-in-USIM': 'SUCI-in-USIM'}),
+
+            Paramtest(param_cls=p13n.SuciRi, val='123',
+                      expect_clean_val='123',
+                      expect_val={'5G-SUCI-RI': '123'}),
+            Paramtest(param_cls=p13n.SuciRi, val='0',
+                      expect_clean_val='0',
+                      expect_val={'5G-SUCI-RI': '0'}),
+            Paramtest(param_cls=p13n.SuciRi, val='9999',
+                      expect_clean_val='9999',
+                      expect_val={'5G-SUCI-RI': '9999'}),
+
+            Paramtest(param_cls=p13n.SuciCalcInfoUe,
+                      val=json.dumps(sucici),
+                      expect_clean_val=sucici,
+                      expect_val={'5G-SUCI-CalcInfo-UE': json.dumps(sucici)}),
+
+            Paramtest(param_cls=p13n.SuciCalcInfoUsim,
+                      val=json.dumps(sucici),
+                      expect_clean_val=sucici,
+                      expect_val={'5G-SUCI-CalcInfo-USIM': json.dumps(sucici)}),
+
+            Paramtest(param_cls=p13n.GfmSuciRi, val='123',
+                      expect_clean_val='123',
+                      expect_val={'GFM-5G-SUCI-RI': '123'}),
+            Paramtest(param_cls=p13n.GfmSuciRi, val='0',
+                      expect_clean_val='0',
+                      expect_val={'GFM-5G-SUCI-RI': '0'}),
+            Paramtest(param_cls=p13n.GfmSuciRi, val='9999',
+                      expect_clean_val='9999',
+                      expect_val={'GFM-5G-SUCI-RI': '9999'}),
+
+            Paramtest(param_cls=p13n.GfmSuciCalcInfoUe,
+                      val=json.dumps(sucici),
+                      expect_clean_val=sucici,
+                      expect_val={'GFM-5G-SUCI-CalcInfo-UE': json.dumps(sucici)}),
+
+        ])
+
+        Paramtest.iff_present_default = False
 
         for sdkey_cls in (
             # thin out the number of tests, as a compromise between completeness and test runtime
@@ -367,7 +468,8 @@ class ConfigurableParameterTest(unittest.TestCase):
 
                 for t in param_tests:
                     test_idx += 1
-                    logloc = f'{upp_fname} {t.param_cls.__name__}(val={valtypestr(t.val)})'
+                    testlog = []
+                    testlog.append(f'{upp_fname} {t.param_cls.__name__}(val={valtypestr(t.val)})')
 
                     param = None
                     try:
@@ -375,21 +477,32 @@ class ConfigurableParameterTest(unittest.TestCase):
                         param.input_value = t.val
                         param.validate()
                     except ValueError as e:
-                        raise ValueError(f'{logloc}: {e}') from e
+                        raise ValueError(f'{" ".join(testlog)}: {e}') from e
 
                     clean_val = param.value
-                    logloc = f'{logloc} clean_val={valtypestr(clean_val)}'
+                    testlog.append(f'clean_val={valtypestr(clean_val)}')
                     if t.expect_clean_val is not None and t.expect_clean_val != clean_val:
-                        raise ValueError(f'{logloc}: expected'
+                        raise ValueError(f'{" ".join(testlog)}: expected'
                                          f' expect_clean_val={valtypestr(t.expect_clean_val)}')
 
                     # on my laptop, deepcopy is about 30% slower than decoding the DER from scratch:
                     # pes = copy.deepcopy(orig_pes)
                     pes = ProfileElementSequence.from_der(der)
+
+                    found = list((t.param_cls.get_value_from_pes(pes) or {}).values())
+                    testlog.append(f"previous value: {found}")
+
+                    if t.iff_present and not found:
+                        testlog.append("skipping, param not in template.")
+                        output = "\nskip: " + "\n  ".join(testlog)
+                        outputs.append(output)
+                        print(output)
+                        continue
+
                     try:
                         param.apply(pes)
                     except ValueError as e:
-                        raise ValueError(f'{logloc} apply_val(clean_val): {e}') from e
+                        raise ValueError(f'{" ".join(testlog)} apply_val(clean_val): {e}') from e
 
                     changed_der = pes.to_der()
 
@@ -407,22 +520,18 @@ class ConfigurableParameterTest(unittest.TestCase):
                     else:
                         read_back_val_type = f'{type(read_back_val).__name__}'
 
-                    logloc = (f'{logloc} read_back_val={valtypestr(read_back_val)}')
+                    testlog.append(f'read_back_val={valtypestr(read_back_val)}')
 
                     if isinstance(read_back_val, dict) and not t.param_cls.get_name() in read_back_val.keys():
-                        raise ValueError(f'{logloc}: expected to find name {t.param_cls.get_name()!r} in read_back_val')
+                        raise ValueError(f'{" ".join(testlog)}: expected to find name {t.param_cls.get_name()!r} in read_back_val')
 
                     expect_val = t.expect_val
                     if not isinstance(expect_val, dict):
                         expect_val = { t.param_cls.get_name(): expect_val }
                     if read_back_val != expect_val:
-                        raise ValueError(f'{logloc}: expected {expect_val=!r}:{type(t.expect_val).__name__}')
+                        raise ValueError(f'{" ".join(testlog)}: expected {expect_val=!r}:{type(t.expect_val).__name__}')
 
-                    ok = logloc.replace(' clean_val', '\n\tclean_val'
-                              ).replace(' read_back_val', '\n\tread_back_val'
-                              ).replace('=', '=\t'
-                              )
-                    output = f'\nok: {ok}'
+                    output = "\nok: " + "\n  ".join(testlog)
                     outputs.append(output)
                     print(output)
 
@@ -558,7 +667,7 @@ class TestEnumParam(unittest.TestCase):
     def test_validate_by_name_exact(self):
         self.assertEqual(p13n.AlgorithmID.validate_val('Milenage'), 1)
         self.assertEqual(p13n.AlgorithmID.validate_val('TUAK'), 2)
-        self.assertEqual(p13n.AlgorithmID.validate_val('usim_test'), 3)
+        self.assertEqual(p13n.AlgorithmID.validate_val('usim-test'), 3)
 
     def test_validate_by_int(self):
         self.assertEqual(p13n.AlgorithmID.validate_val(1), 1)
@@ -571,7 +680,6 @@ class TestEnumParam(unittest.TestCase):
         self.assertEqual(p13n.AlgorithmID.validate_val('tuak'), 2)
 
     def test_validate_fuzzy_hyphen_underscore(self):
-        # 'usim-test' has a hyphen; enum member is 'usim_test' — must fuzzy-match
         self.assertEqual(p13n.AlgorithmID.validate_val('usim-test'), 3)
 
     def test_validate_invalid_name(self):
@@ -608,7 +716,7 @@ class TestEnumParam(unittest.TestCase):
     def test_map_val_known(self):
         self.assertEqual(p13n.AlgorithmID.map_val_to_name(1), 'Milenage')
         self.assertEqual(p13n.AlgorithmID.map_val_to_name(2), 'TUAK')
-        self.assertEqual(p13n.AlgorithmID.map_val_to_name(3), 'usim_test')
+        self.assertEqual(p13n.AlgorithmID.map_val_to_name(3), 'usim-test')
 
     def test_map_val_unknown_nonstrict(self):
         self.assertIsNone(p13n.AlgorithmID.map_val_to_name(99))
@@ -622,7 +730,9 @@ class TestEnumParam(unittest.TestCase):
     def test_name_normalize(self):
         self.assertEqual(p13n.AlgorithmID.name_normalize('Milenage'), 'Milenage')
         self.assertEqual(p13n.AlgorithmID.name_normalize('milenage'), 'Milenage')
-        self.assertEqual(p13n.AlgorithmID.name_normalize('usim-test'), 'usim_test')
+        self.assertEqual(p13n.AlgorithmID.name_normalize('usimtest'), 'usim-test')
+        self.assertEqual(p13n.AlgorithmID.name_normalize('usim_test'), 'usim-test')
+        self.assertEqual(p13n.AlgorithmID.name_normalize('USIM Test'), 'usim-test')
 
     # --- clean_name_str ---
 

tests/unittests/test_globalplatform.py

@@ -17,10 +17,7 @@
 
 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 *
@@ -286,41 +283,6 @@ 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'))
@@ -328,208 +290,6 @@ 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')
@@ -538,180 +298,5 @@ 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,17 +37,6 @@ 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.