WIP: saip.data_source

Change-Id: Ia6f70ff467ba58024d94742ba5cecd8141b93ad6

pySim/esim/saip/data_source.py

@@ -0,0 +1,90 @@
+# Data sources: Provding data for profile personalization
+#
+# (C) 2024 by Harald Welte <laforge@osmocom.org>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import abc
+import secrets
+
+from Cryptodome.Random import get_random_bytes
+
+class DataSource(abc.ABC):
+    """Base class for something that can provide data during a personalization process."""
+
+    @abc.abstractmethod
+    def generate_one(self):
+        pass
+
+
+class DataSourceFixed(DataSource):
+    """A data source that provides a fixed value (of any type).
+
+    Parameters:
+        fixed_value: The fixed value that shall be used during each data generation
+    """
+    def __init__(self, fixed_value, **kwargs):
+        self.fixed_value = fixed_value
+        super().__init__(**kwargs)
+
+    def generate_one(self):
+        return self.fixed_value
+
+
+class DataSourceIncrementing(DataSource):
+    """A data source that provides incrementing integer numbers.
+
+    Parameters:
+        base_value: The start value (value returned during first data generation)
+        step_size: Increment step size (Default: 1)
+    """
+    def __init__(self, base_value: int, **kwargs):
+        self.base_value = int(base_value)
+        self.step_size = kwargs.pop('step_size', 1)
+        self.i = 0
+        super().__init__(**kwargs)
+
+    def generate_one(self):
+        val = self.base_value + self.i
+        self.i += self.step_size
+        return val
+
+
+class DataSourceRandomBytes(DataSource):
+    """A data source that provides a configurable number of random bytes.
+
+    Parameters:
+        size: Number of bytes to generate each turn
+    """
+    def __init__(self, size: int, **kwargs):
+        self.size = size
+        super().__init__(**kwargs)
+
+    def generate_one(self):
+        return get_random_bytes(self.size)
+
+
+class DataSourceRandomUInt(DataSource):
+    """A data source that provides a configurable unsigned integer value.
+
+    Parameters:
+        below: Number one greater than the maximum permitted random unsigned integer
+    """
+    def __init__(self, below: int, **kwargs):
+        self.below = below
+        super().__init__(**kwargs)
+
+    def generate_one(self):
+        return secrets.randbelow(self.below)
+

pySim/global_platform/__init__.py

@@ -17,6 +17,7 @@ You should have received a copy of the GNU General Public License
 along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 
+import io
 from copy import deepcopy
 from typing import Optional, List, Dict, Tuple
 from construct import Optional as COptional
@@ -434,6 +435,23 @@ class GpRegistryRelatedData(BER_TLV_IE, tag=0xe3, nested=[ApplicationAID, LifeCy
                                                           ExecutableModuleAID, AssociatedSecurityDomainAID]):
     pass
 
+
+# Section 11.6.2.3 / Table 11-58
+class SecurityDomainAid(BER_TLV_IE, tag=0x4f):
+    _construct = GreedyBytes
+class LoadFileDataBlockSignature(BER_TLV_IE, tag=0xc3):
+    _construct = GreedyBytes
+class DapBlock(BER_TLV_IE, tag=0xe2, nested=[SecurityDomainAid, LoadFileDataBlockSignature]):
+    pass
+class LoadFileDataBlock(BER_TLV_IE, tag=0xc4):
+    _construct = GreedyBytes
+class Icv(BER_TLV_IE, tag=0xd3):
+    _construct = GreedyBytes
+class CipheredLoadFileDataBlock(BER_TLV_IE, tag=0xd4):
+    _construct = GreedyBytes
+class LoadFile(TLV_IE_Collection, nested=[DapBlock, LoadFileDataBlock, Icv, CipheredLoadFileDataBlock]):
+    pass
+
 # Application Dedicated File of a Security Domain
 class ADF_SD(CardADF):
     StoreData = BitStruct('last_block'/Flag,
@@ -675,6 +693,31 @@ class ADF_SD(CardADF):
             ifi_bytes = build_construct(InstallForInstallCD, decoded)
             self.install(p1, 0x00, b2h(ifi_bytes))
 
+        inst_load_parser = argparse.ArgumentParser()
+        inst_load_parser.add_argument('--load-file-aid', type=is_hexstr, required=True,
+                                      help='AID of the loded file')
+        inst_load_parser.add_argument('--security-domain-aid', type=is_hexstr, default='',
+                                      help='AID of the Security Domain into which the file shalle be added')
+        inst_load_parser.add_argument('--load-file-hash', type=is_hexstr, default='',
+                                      help='Load File Data Block Hash')
+        inst_inst_parser.add_argument('--load-parameters', type=is_hexstr, default='',
+                                      help='Load Token (Section GPCS C.4.1)')
+        inst_inst_parser.add_argument('--load-token', type=is_hexstr, default='',
+                                      help='Load Token (Section GPCS C.4.1)')
+
+        @cmd2.with_argparser(inst_load_parser)
+        def do_install_for_load(self, opts):
+            """Perform GlobalPlatform INSTALL [for load] command."""
+            if opts.load_token != '' and opts.load_file_hash == '':
+                raise ValueError('Load File Data Block Hash is mandatory if a Load Token is present')
+            InstallForLoadCD = Struct('load_file_aid'/HexAdapter(Prefixed(Int8ub, GreedyBytes)),
+                                      'security_domain_aid'/HexAdapter(Prefixed(Int8ub, GreedyBytes)),
+                                      'load_file_hash'/HexAdapter(Prefixed(Int8ub, GreedyBytes)),
+                                      'load_parameters'/HexAdapter(Prefixed(Int8ub, GreedyBytes)),
+                                      'load_token'/HexAdapter(Prefixed(Int8ub, GreedyBytes)))
+            ifl_bytes = build_construct(InstallForLoadCD, vars(opts))
+            self.install(0x02, 0x00, b2h(ifl_bytes))
+
         def install(self, p1:int, p2:int, data:Hexstr) -> ResTuple:
             cmd_hex = "80E6%02x%02x%02x%s" % (p1, p2, len(data)//2, data)
             return self._cmd.lchan.scc.send_apdu_checksw(cmd_hex)
@@ -718,6 +761,37 @@ class ADF_SD(CardADF):
             cmd_hex = "80E4%02x%02x%02x%s" % (p1, p2, len(data)//2, data)
             return self._cmd.lchan.scc.send_apdu_checksw(cmd_hex)
 
+        load_parser = argparse.ArgumentParser()
+        # we make this a required --optional argument now, so we can later have other sources for load data
+        load_parser.add_argument('--from-file', required=True)
+
+        @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."""
+            with open(opts.from_file, 'rb') as f:
+                self.load(f)
+
+        def load(self, stream: io.RawIOBase, chunk_len:int = 240):
+            # we might want to tune chunk_len based on the overhead of the used SCP?
+            contents = stream.readall()
+            # build TLV according to 11.6.2.3 / Table 11-58 for unencrypted case
+            remainder = b'\xC4' + bertlv_encode_len(len(contents)) + contents
+            # transfer this in vaious chunks to the card
+            total_size = len(remainder)
+            block_nr = 0
+            while len(remainder):
+                block = remainder[:chunk_len]
+                remainder = remainder[chunk_len:]
+                # build LOAD command APDU according to 11.6.2 / Table 11-56
+                p1 = 0x80 if len(remainder) else 0x00
+                p2 = block_nr % 256
+                block_nr += 1
+                cmd_hex = "80E8%02x%02x%02x%s" % (p1, p2, len(block), b2h(block))
+                _rsp_hex, _sw = self._cmd.lchan.scc.send_apdu_checksw(cmd_hex)
+            self._cmd.poutput("Loaded a total of %u bytes in %u blocks. Don't forget install_for_load now!" % (total_size, block_nr))
+
+
         est_scp02_parser = argparse.ArgumentParser()
         est_scp02_parser.add_argument('--key-ver', type=auto_uint8, required=True,
                                       help='Key Version Number (KVN)')

saip-test.py

@@ -0,0 +1,100 @@
+#!/usr/bin/env python3
+
+from pySim.utils import b2h, h2b
+from pySim.esim.saip import *
+from pySim.esim.saip.validation import *
+
+from pySim.pprint import HexBytesPrettyPrinter
+
+pp = HexBytesPrettyPrinter(indent=4,width=500)
+
+import abc
+
+
+
+
+with open('smdpp-data/upp/TS48v2_SAIP2.3_NoBERTLV.der', 'rb') as f:
+    pes = ProfileElementSequence.from_der(f.read())
+
+if False:
+    # iterate over each pe in the pes.pe_list
+    for pe in pes.pe_list:
+        print("="*70 + " " + pe.type)
+        pp.pprint(pe.decoded)
+
+if False:
+    # sort by PE type and show all PE within that type
+    for pe_type in pes.pe_by_type.keys():
+        print("="*70 + " " + pe_type)
+        for pe in pes.pe_by_type[pe_type]:
+            pp.pprint(pe)
+            pp.pprint(pe.decoded)
+
+checker = CheckBasicStructure()
+checker.check(pes)
+
+if False:
+    for naa in pes.pes_by_naa:
+        i = 0
+        for naa_instance in pes.pes_by_naa[naa]:
+            print("="*70 + " " + naa + str(i))
+            i += 1
+            for pe in naa_instance:
+                pp.pprint(pe.type)
+                for d in pe.decoded:
+                    print("    %s" % d)
+                    #pp.pprint(pe.decoded[d])
+                #if pe.type in ['akaParameter', 'pinCodes', 'pukCodes']:
+                #    pp.pprint(pe.decoded)
+
+
+from pySim.esim.saip.personalization import *
+
+params = [Iccid('984944000000000000'), Imsi('901990123456789'),
+          Puk1(value='01234567'), Puk2(value='98765432'), Pin1('1111'), Pin2('2222'), Adm1('11111111'),
+          K(h2b('000102030405060708090a0b0c0d0e0f')), Opc(h2b('101112131415161718191a1b1c1d1e1f')),
+          SdKeyScp80_01Kic(h2b('000102030405060708090a0b0c0d0e0f'))]
+
+from pySim.esim.saip.templates import *
+
+for p in params:
+    p.apply(pes)
+
+if False:
+    for pe in pes:
+        pp.pprint(pe.decoded)
+        pass
+
+if True:
+    naas = pes.pes_by_naa.keys()
+    for naa in naas:
+        for pe in pes.pes_by_naa[naa][0]:
+            print(pe)
+            #pp.pprint(pe.decoded)
+            #print(pe.header)
+            tpl_id = pe.templateID
+            if tpl_id:
+                prof = ProfileTemplateRegistry.get_by_oid(tpl_id)
+                print(prof)
+                #pp.pprint(pe.decoded)
+            for fname, fdata in pe.files.items():
+                print()
+                print("============== %s" % fname)
+                ftempl = None
+                if prof:
+                    ftempl = prof.files_by_pename[fname]
+                print("Template: %s" % repr(ftempl))
+                print("Data: %s" % fdata)
+                file = File(fname, fdata, ftempl)
+                print(repr(file))
+                #pp.pprint(pe.files)
+
+if True:
+    # iterate over each pe in the pes (using its __iter__ method)
+    for pe in pes:
+        print("="*70 + " " + pe.type)
+        pp.pprint(pe.decoded)
+
+
+
+#print(ProfileTemplateRegistry.by_oid)

setup.py

@@ -3,7 +3,8 @@ from setuptools import setup
 setup(
     name='pySim',
     version='1.0',
-    packages=['pySim', 'pySim.legacy', 'pySim.transport', 'pySim.apdu', 'pySim.apdu_source'],
+    packages=['pySim', 'pySim.legacy', 'pySim.transport', 'pySim.apdu', 'pySim.apdu_source',
+              'pySim.esim'],
     url='https://osmocom.org/projects/pysim/wiki',
     license='GPLv2',
     author_email='simtrace@lists.osmocom.org',

tests/test_esim_saip.py

@@ -21,6 +21,7 @@ import copy
 
 from pySim.utils import h2b, b2h
 from pySim.esim.saip import *
+from pySim.esim.saip.data_source import *
 from pySim.esim.saip.personalization import *
 from pprint import pprint as pp
 
@@ -64,5 +65,47 @@ class SaipTest(unittest.TestCase):
         pes.to_der()
 
 
+class DataSourceTest(unittest.TestCase):
+    def test_fixed(self):
+        FIXED = b'\x01\x02\x03'
+        ds = DataSourceFixed(FIXED)
+        self.assertEqual(ds.generate_one(), FIXED)
+        self.assertEqual(ds.generate_one(), FIXED)
+        self.assertEqual(ds.generate_one(), FIXED)
+
+    def test_incrementing(self):
+        BASE_VALUE = 100
+        ds = DataSourceIncrementing(BASE_VALUE)
+        self.assertEqual(ds.generate_one(), BASE_VALUE)
+        self.assertEqual(ds.generate_one(), BASE_VALUE+1)
+        self.assertEqual(ds.generate_one(), BASE_VALUE+2)
+        self.assertEqual(ds.generate_one(), BASE_VALUE+3)
+
+    def test_incrementing_step3(self):
+        BASE_VALUE = 300
+        ds = DataSourceIncrementing(BASE_VALUE, step_size=3)
+        self.assertEqual(ds.generate_one(), BASE_VALUE)
+        self.assertEqual(ds.generate_one(), BASE_VALUE+3)
+        self.assertEqual(ds.generate_one(), BASE_VALUE+6)
+
+    def test_random(self):
+        ds = DataSourceRandomBytes(8)
+        res = []
+        for i in range(0,100):
+            res.append(ds.generate_one())
+        for r in res:
+            self.assertEqual(len(r), 8)
+        # ensure no duplicates exist
+        self.assertEqual(len(set(res)), len(res))
+
+    def test_random_int(self):
+        ds = DataSourceRandomUInt(below=256)
+        res = []
+        for i in range(0,100):
+            res.append(ds.generate_one())
+        for r in res:
+            self.assertTrue(r < 256)
+
+
 if __name__ == "__main__":
 	unittest.main()