ConfigurableParameter.get_typical_input_len: limit to 10 lines

Change-Id: Ia3d79e786f397a02bf2a8fafac5030d1198d9f76

README.md

@@ -97,7 +97,7 @@ Please install the following dependencies:
  - pyscard
  - pyserial
  - pytlv
- - pyyaml >= 5.1
+ - pyyaml >= 5.4
  - smpp.pdu (from `github.com/hologram-io/smpp.pdu`)
  - termcolor
 

pySim-prog.py

@@ -27,7 +27,6 @@
 import hashlib
 import argparse
 import os
-import random
 import re
 import sys
 import traceback
@@ -436,7 +435,7 @@ def gen_parameters(opts):
         if not re.match('^[0-9a-fA-F]{32}$', ki):
             raise ValueError('Ki needs to be 128 bits, in hex format')
     else:
-        ki = ''.join(['%02x' % random.randrange(0, 256) for i in range(16)])
+        ki = os.urandom(16).hex()
 
     # OPC (random)
     if opts.opc is not None:
@@ -447,7 +446,7 @@ def gen_parameters(opts):
     elif opts.op is not None:
         opc = derive_milenage_opc(ki, opts.op)
     else:
-        opc = ''.join(['%02x' % random.randrange(0, 256) for i in range(16)])
+        opc = os.urandom(16).hex()
 
     pin_adm = sanitize_pin_adm(opts.pin_adm, opts.pin_adm_hex)
 

pySim/esim/saip/personalization.py

@@ -20,13 +20,14 @@ import io
 import os
 import re
 import pprint
+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
 from pySim.ts_51_011 import EF_SMSP
 from pySim.esim.saip import param_source
 from pySim.esim.saip import ProfileElement, ProfileElementSD, ProfileElementSequence
@@ -291,7 +292,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):
@@ -618,10 +621,28 @@ class SmspTpScAddr(ConfigurableParameter):
             ef_smsp_dec['tp_sc_addr']['ton_npi']['type_of_number'] = 'international' if international else 'unknown'
             # ensure the parameter_indicators.tp_sc_addr is True
             ef_smsp_dec['parameter_indicators']['tp_sc_addr'] = True
+
+            # alpha_id padding: to make room for a human readable SMSC name that can be provisioned to the profile later
+            # on, alpha_id needs to be empty but padded 0xff to some length.
+            # - alpha_id is optional, setting alpha_id = '' ensures the IE is present.
+            # - the length of the file is 28+Y where Y is the length of the alpha_id -- here the intended length of our padding
+            #   (see 3GPP TS 31.102 4.2.27 EF.SMSP). So if we want a maximum length of alpha_id = 14, we set the total
+            #   file size to 28+14 = 42.
+            # - this file size has to go in two places: encode_record_bin() needs to know the length to encode the right
+            #   length of fillFileContent.
+            # - the f_smsp needs to show the right file size in the PES, as in
+            #      'ef-smsp': [('fileDescriptor', {'efFileSize': '2a', ...
+            #   (where 2a == 42)
+            # - 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'] = ''
-            # re-encode into the File body
+
-            f_smsp.body = ef_smsp.encode_record_bin(ef_smsp_dec, 1)
+            # we can set this to choose a fixed length:
-            #print("SMSP  (new): %s" % f_smsp.body)
+            #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.
+            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)
 
@@ -1176,3 +1197,184 @@ class TuakNumberOfKeccak(IntegerParam, AlgoConfig):
     max_val = 255
     example_input = '1'
     default_source = param_source.ConstantSource
+
+
+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-on': True, 'SUCI-off': False }
+    example_input = sorted(value_map.keys())[0]
+
+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 = sorted(value_map.keys())[0]
+
+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 SuciCalcInfo(ConfigurableParameter):
+    """SUCI Calculation Information as in section 4.4.11.8 of 3GPP TS 31.102"""
+    name = '5G-SUCI-CalcInfo'
+    example_input = '{}'
+    default_source = param_source.ConstantSource
+    allow_types = (str,)
+    max_len = 2000
+
+    @classmethod
+    def validate_val(cls, val):
+        val = super().validate_val(val)
+
+        if not val:
+            raise ValueError("SUCI Calc Info value is empty -- should at least be an empty dict like '{}'")
+
+        # 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 (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()
+            f_sucici.body = ef_sucici.encode_bin(val)
+            pe.file2pe(f_sucici)
+
+    @classmethod
+    def apply_val(cls, pes: ProfileElementSequence, val):
+        cls._apply_suci(pes, val, "df-5gs", "ef-suci-calc-info")
+        cls._apply_suci(pes, val, "df-saip", "ef-suci-calc-info-usim")
+
+    @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)
+            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
+
+            yield { cls.name: json.dumps(sucici) }
+
+    @classmethod
+    def get_values_from_pes(cls, pes: ProfileElementSequence):
+        yield from cls._get_suci(pes, "df-5gs", "ef-suci-calc-info")
+        yield from cls._get_suci(pes, "df-saip", "ef-suci-calc-info-usim")

pySim/ts_31_102.py

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

requirements.txt

@@ -6,7 +6,7 @@ jsonpath-ng
 construct>=2.10.70
 bidict
 pyosmocom>=0.0.12
-pyyaml>=5.1
+pyyaml>=5.4
 termcolor
 colorlog
 pycryptodomex

setup.py

@@ -26,7 +26,7 @@ setup(
         "construct >= 2.10.70",
         "bidict",
         "pyosmocom >= 0.0.12",
-        "pyyaml >= 5.1",
+        "pyyaml >= 5.4",
         "termcolor",
         "colorlog",
         "pycryptodomex",

tests/unittests/test_configurable_parameters.py

@@ -21,6 +21,7 @@ import io
 import sys
 import unittest
 import io
+import json
 from importlib import resources
 from osmocom.utils import hexstr
 from pySim.esim.saip import ProfileElementSequence
@@ -60,11 +61,15 @@ class ConfigurableParameterTest(unittest.TestCase):
                )
 
         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',
@@ -276,8 +281,56 @@ class ConfigurableParameterTest(unittest.TestCase):
                       val=3,
                       expect_clean_val=3,
                       expect_val='3'),
+        ]
 
-            ]
+        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.SuciCalcInfo,
+                      val=json.dumps(sucici),
+                      expect_clean_val=sucici,
+                      expect_val={'5G-SUCI-CalcInfo': 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
@@ -373,7 +426,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:
@@ -381,21 +435,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()
 
@@ -413,22 +478,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'
+                    output = "\nok: " + "\n  ".join(testlog)
-                              ).replace(' read_back_val', '\n\tread_back_val'
-                              ).replace('=', '=\t'
-                              )
-                    output = f'\nok: {ok}'
                     outputs.append(output)
                     print(output)