Compare commits

..

1 Commits

Author SHA1 Message Date
catarrh 1b2fc595ce Allow saving unknown files with raw data
When a file has no custom encoder, the decode path returns {'raw': ...}
but the encode path raised NotImplementedError. Add a fallback in all
encoder methods to return the raw data directly when present in the
abstract_data dict, allowing unknown files to round-trip.
2026-07-04 11:48:53 +03:00
10 changed files with 2909 additions and 3717 deletions
+38 -49
View File
@@ -300,51 +300,6 @@ class ADF_ARAM(CardADF):
'major': v_major, 'minor': v_minor, 'patch': v_patch}}]) 'major': v_major, 'minor': v_minor, 'patch': v_patch}}])
return ADF_ARAM.xceive_apdu_tlv(scc, '80cadf21', cmd_do, ResponseAramConfigDO) return ADF_ARAM.xceive_apdu_tlv(scc, '80cadf21', cmd_do, ResponseAramConfigDO)
@staticmethod
def store_ref_ar_do(scc, aid:Hexstr, aid_empty:bool, device_app_id:Hexstr, pkg_ref:str,
apdu_filter:Hexstr, apdu_never:bool, apdu_always:bool,
nfc_always:bool, nfc_never:bool, android_permissions:Hexstr):
# REF
ref_do_content = []
if aid is not None:
ref_do_content += [{'aid_ref_do': aid}]
elif aid_empty:
ref_do_content += [{'aid_ref_empty_do': None}]
ref_do_content += [{'dev_app_id_ref_do': device_app_id}]
if pkg_ref:
ref_do_content += [{'pkg_ref_do': {'package_name_string': pkg_ref}}]
# AR
ar_do_content = []
if apdu_never:
ar_do_content += [{'apdu_ar_do': {'generic_access_rule': 'never'}}]
elif apdu_always:
ar_do_content += [{'apdu_ar_do': {'generic_access_rule': 'always'}}]
elif apdu_filter:
if len(apdu_filter) % 16:
raise ValueError(f'Invalid non-modulo-16 length of APDU filter: {len(apdu_filter)}')
offset = 0
apdu_filter_list = []
while offset < len(apdu_filter):
apdu_filter_list += [{'header': apdu_filter[offset:offset+8],
'mask': apdu_filter[offset+8:offset+16]}]
offset += 16 # Move offset to the beginning of the next apdu_filter object
ar_do_content += [{'apdu_ar_do': {'apdu_filter': apdu_filter_list}}]
if nfc_never:
ar_do_content += [{'nfc_ar_do': {'nfc_event_access_rule': 'never'}}]
elif nfc_always:
ar_do_content += [{'nfc_ar_do': {'nfc_event_access_rule': 'always'}}]
if android_permissions:
ar_do_content += [{'perm_ar_do': {'permissions': android_permissions}}]
d = [{'ref_ar_do': [{'ref_do': ref_do_content}, {'ar_do': ar_do_content}]}]
csrado = CommandStoreRefArDO()
csrado.from_val_dict(d)
return ADF_ARAM.store_data(scc, csrado)
@staticmethod
def aram_delete_all(scc):
deldo = CommandDelete()
return ADF_ARAM.store_data(scc, deldo)
@with_default_category('Application-Specific Commands') @with_default_category('Application-Specific Commands')
class AddlShellCommands(CommandSet): class AddlShellCommands(CommandSet):
def do_aram_get_all(self, _opts): def do_aram_get_all(self, _opts):
@@ -389,15 +344,48 @@ class ADF_ARAM(CardADF):
@cmd2.with_argparser(store_ref_ar_do_parse) @cmd2.with_argparser(store_ref_ar_do_parse)
def do_aram_store_ref_ar_do(self, opts): def do_aram_store_ref_ar_do(self, opts):
"""Perform STORE DATA [Command-Store-REF-AR-DO] to store a (new) access rule.""" """Perform STORE DATA [Command-Store-REF-AR-DO] to store a (new) access rule."""
res_do = ADF_ARAM.store_ref_ar_do(self._cmd.lchan.scc, opts.aid, opts.aid_empty, opts.device_app_id, # REF
opts.pkg_ref, opts.apdu_filter, opts.apdu_never, opts.apdu_always, ref_do_content = []
opts.nfc_always, opts.nfc_never, opts.android_permissions) if opts.aid is not None:
ref_do_content += [{'aid_ref_do': opts.aid}]
elif opts.aid_empty:
ref_do_content += [{'aid_ref_empty_do': None}]
ref_do_content += [{'dev_app_id_ref_do': opts.device_app_id}]
if opts.pkg_ref:
ref_do_content += [{'pkg_ref_do': {'package_name_string': opts.pkg_ref}}]
# AR
ar_do_content = []
if opts.apdu_never:
ar_do_content += [{'apdu_ar_do': {'generic_access_rule': 'never'}}]
elif opts.apdu_always:
ar_do_content += [{'apdu_ar_do': {'generic_access_rule': 'always'}}]
elif opts.apdu_filter:
if len(opts.apdu_filter) % 16:
raise ValueError(f'Invalid non-modulo-16 length of APDU filter: {len(opts.apdu_filter)}')
offset = 0
apdu_filter = []
while offset < len(opts.apdu_filter):
apdu_filter += [{'header': opts.apdu_filter[offset:offset+8],
'mask': opts.apdu_filter[offset+8:offset+16]}]
offset += 16 # Move offset to the beginning of the next apdu_filter object
ar_do_content += [{'apdu_ar_do': {'apdu_filter': apdu_filter}}]
if opts.nfc_always:
ar_do_content += [{'nfc_ar_do': {'nfc_event_access_rule': 'always'}}]
elif opts.nfc_never:
ar_do_content += [{'nfc_ar_do': {'nfc_event_access_rule': 'never'}}]
if opts.android_permissions:
ar_do_content += [{'perm_ar_do': {'permissions': opts.android_permissions}}]
d = [{'ref_ar_do': [{'ref_do': ref_do_content}, {'ar_do': ar_do_content}]}]
csrado = CommandStoreRefArDO()
csrado.from_val_dict(d)
res_do = ADF_ARAM.store_data(self._cmd.lchan.scc, csrado)
if res_do: if res_do:
self._cmd.poutput_json(res_do.to_dict()) self._cmd.poutput_json(res_do.to_dict())
def do_aram_delete_all(self, _opts): def do_aram_delete_all(self, _opts):
"""Perform STORE DATA [Command-Delete[all]] to delete all access rules.""" """Perform STORE DATA [Command-Delete[all]] to delete all access rules."""
res_do = ADF_ARAM.aram_delete_all(self._cmd.lchan.scc) deldo = CommandDelete()
res_do = ADF_ARAM.store_data(self._cmd.lchan.scc, deldo)
if res_do: if res_do:
self._cmd.poutput_json(res_do.to_dict()) self._cmd.poutput_json(res_do.to_dict())
@@ -406,6 +394,7 @@ class ADF_ARAM(CardADF):
(Proprietary feature that is specific to sysmocom's fork of Bertrand Martels ARA-M implementation.)""" (Proprietary feature that is specific to sysmocom's fork of Bertrand Martels ARA-M implementation.)"""
self._cmd.lchan.scc.send_apdu_checksw('80e2900001A1', '9000') self._cmd.lchan.scc.send_apdu_checksw('80e2900001A1', '9000')
# SEAC v1.1 Section 4.1.2.2 + 5.1.2.2 # SEAC v1.1 Section 4.1.2.2 + 5.1.2.2
sw_aram = { sw_aram = {
'ARA-M': { 'ARA-M': {
+2 -12
View File
@@ -411,30 +411,20 @@ class JsonHttpApiClient():
# SGP.22, section 6.5.1) # SGP.22, section 6.5.1)
if response.status_code != self.api_func.expected_http_status: if response.status_code != self.api_func.expected_http_status:
raise HttpStatusError(response) 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) raise HttpHeaderError(response)
if not response.headers.get('X-Admin-Protocol', 'gsma/rsp/v2.unknown').startswith('gsma/rsp/v2.'): if not response.headers.get('X-Admin-Protocol', 'gsma/rsp/v2.unknown').startswith('gsma/rsp/v2.'):
raise HttpHeaderError(response) raise HttpHeaderError(response)
# Decode response and return the result back to the caller # Decode response and return the result back to the caller
if response.content: if response.content:
if resp_content_type.startswith('application/json'): output = self.api_func.decode_client(response.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 # 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) # (the presence of the header field is checked by the decode_client method)
if 'header' in output: if 'header' in output:
if output['header']['functionExecutionStatus']['status'] not in ['Executed-Success','Executed-WithWarning']: if output['header']['functionExecutionStatus']['status'] not in ['Executed-Success','Executed-WithWarning']:
raise ApiError(output['header']['functionExecutionStatus']) raise ApiError(output['header']['functionExecutionStatus'])
return output return output
return None return None
class JsonHttpApiServer(): class JsonHttpApiServer():
+4 -56
View File
@@ -34,7 +34,7 @@ from pySim import ts_102_222
from pySim.utils import dec_imsi from pySim.utils import dec_imsi
from pySim.ts_102_221 import FileDescriptor from pySim.ts_102_221 import FileDescriptor
from pySim.filesystem import CardADF, Path from pySim.filesystem import CardADF, Path
from pySim.ts_31_102 import ADF_USIM, EF_UST, EF_SUCI_Calc_Info from pySim.ts_31_102 import ADF_USIM
from pySim.ts_31_103 import ADF_ISIM from pySim.ts_31_103 import ADF_ISIM
from pySim.esim import compile_asn1_subdir from pySim.esim import compile_asn1_subdir
from pySim.esim.saip import templates from pySim.esim.saip import templates
@@ -1517,11 +1517,8 @@ class ProfileElementHeader(ProfileElement):
def mandatory_service_add(self, service_name): def mandatory_service_add(self, service_name):
self.decoded['eUICC-Mandatory-services'][service_name] = None 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): def mandatory_service_remove(self, service_name):
if self.mandatory_service_present(service_name): if service_name in self.decoded['eUICC-Mandatory-services'].keys():
del self.decoded['eUICC-Mandatory-services'][service_name] del self.decoded['eUICC-Mandatory-services'][service_name]
else: else:
raise ValueError("service not in eUICC-Mandatory-services list, cannot remove") raise ValueError("service not in eUICC-Mandatory-services list, cannot remove")
@@ -1729,61 +1726,12 @@ class ProfileElementSequence:
if 'BT' in ftype_list: if 'BT' in ftype_list:
svc_set.add('ber-tlv') svc_set.add('ber-tlv')
# FIXME:dfLinked files (scan all files, check for non-empty Fcp.linkPath presence of DFs) # FIXME:dfLinked files (scan all files, check for non-empty Fcp.linkPath presence of DFs)
# TODO: 5G related bits (derive from EF.UST or file presence?)
# 5G:
# - When SUCI is:
# - enabled (EF.UST 124 = true)
# AND
# - calculated in the USIM (EF.UST 125 = true),
# then eUICC-Mandatory-services needs 'get-identity'.
# - 'get-identity' implies that the eUICC must support ONE OF profile-A OR profile-B.
# (One might assume from this that, when SUCI-CalcInfo for USIM in DF.SAIP contains both key types, then no
# profile-A or B services need to be requested explicitly. However, the correct logic is:)
# - Iff the SUCI-CalcInfo for USIM (DF.SAIP) contains a key of profile-A ("identifier": 1),
# then eUICC-Mandatory-services needs 'profile-a-x25519'.
# - Same: profile-B ("identifier": 2) needs 'profile-b-p256'.
# - (When SUCI is calculated in the UE, then the eUICC does not need to provide any of these services.)
suci_in_usim_enabled = False
try:
f_ust = self.get_pe_for_type("usim").files["ef-ust"]
ust = EF_UST().decode_bin(f_ust.body)
suci_in_usim_enabled = ust[124]['activated'] and ust[125]['activated']
except (KeyError, AttributeError):
pass
if suci_in_usim_enabled:
svc_set.add('get-identity')
# now check for profile-a and profile-b presence
suci_calcinfo_has_profile_a = False
suci_calcinfo_has_profile_b = False
try:
f_sucici = self.get_pe_for_type("df-saip").files["ef-suci-calc-info-usim"]
sucici = EF_SUCI_Calc_Info().decode_bin(f_sucici.body) or {}
for prot_scheme in sucici['prot_scheme_id_list']:
if not isinstance(prot_scheme, dict):
continue
ps_id = prot_scheme["identifier"]
if ps_id == 1:
suci_calcinfo_has_profile_a = True
elif ps_id == 2:
suci_calcinfo_has_profile_b = True
except (KeyError, AttributeError):
pass
if suci_calcinfo_has_profile_a:
# The profile has a profile-A key, so require that
svc_set.add('profile-a-x25519')
if suci_calcinfo_has_profile_b:
# The profile has a profile-B key, so require that
svc_set.add('profile-b-p256')
hdr_pe = self.get_pe_for_type('header') hdr_pe = self.get_pe_for_type('header')
# patch in the 'manual' services from the existing list: # patch in the 'manual' services from the existing list:
old_svc_set = set()
for old_svc in hdr_pe.decoded['eUICC-Mandatory-services'].keys(): for old_svc in hdr_pe.decoded['eUICC-Mandatory-services'].keys():
if old_svc in manual_services: if old_svc in manual_services:
old_svc_set.add(old_svc) 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} hdr_pe.decoded['eUICC-Mandatory-services'] = {x: None for x in svc_set}
def rebuild_mandatory_gfstelist(self): def rebuild_mandatory_gfstelist(self):
+1 -15
View File
@@ -20,9 +20,6 @@
import copy import copy
import pprint import pprint
import logging
import traceback
import inspect
from typing import Generator, Union from typing import Generator, Union
from pySim.esim.saip.personalization import ConfigurableParameter from pySim.esim.saip.personalization import ConfigurableParameter
from pySim.esim.saip import param_source from pySim.esim.saip import param_source
@@ -30,10 +27,6 @@ from pySim.esim.saip import ProfileElementSequence, ProfileElementSD
from pySim.global_platform import KeyUsageQualifier from pySim.global_platform import KeyUsageQualifier
from osmocom.utils import b2h 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 # a list of ConfigurableParameter classes and/or ConfigurableParameter class instances
ParamList = list[Union[type[ConfigurableParameter], ConfigurableParameter]] ParamList = list[Union[type[ConfigurableParameter], ConfigurableParameter]]
@@ -128,12 +121,8 @@ class BatchPersonalization:
value = p.param_cls.validate_val(input_value) value = p.param_cls.validate_val(input_value)
p.param_cls.apply_val(pes, value) p.param_cls.apply_val(pes, value)
except Exception as e: 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 raise ValueError(f'{p.param_cls.get_name()} fed by {p.src.name}: {e}') from e
pes.rebuild_mandatory_services()
yield pes yield pes
@@ -333,15 +322,12 @@ class BatchAudit(list):
return batch_audit return batch_audit
def to_csv_rows(self, headers=True, sort_key=None, column_blacklist=None): def to_csv_rows(self, headers=True, sort_key=None):
"""generator that yields all audits' values as rows, useful feed to a csv.writer.""" """generator that yields all audits' values as rows, useful feed to a csv.writer."""
columns = set() columns = set()
for audit in self: for audit in self:
columns.update(audit.keys()) columns.update(audit.keys())
if column_blacklist:
columns.difference_update(set(column_blacklist))
columns = tuple(sorted(columns, key=sort_key)) columns = tuple(sorted(columns, key=sort_key))
if headers: if headers:
+59 -535
View File
@@ -19,26 +19,17 @@ import abc
import enum import enum
import io import io
import re import re
import json
from typing import List, Tuple, Generator, Optional from typing import List, Tuple, Generator, Optional
from construct.core import StreamError
from osmocom.tlv import camel_to_snake from osmocom.tlv import camel_to_snake
from osmocom.utils import hexstr from osmocom.utils import hexstr
from pySim.utils import enc_iccid, dec_iccid, enc_imsi, dec_imsi, h2b, b2h, rpad, sanitize_iccid 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, EF_UST, EF_Routing_Indicator, EF_SUCI_Calc_Info, DF_USIM_5GS
from pySim.ts_51_011 import EF_SMSP from pySim.ts_51_011 import EF_SMSP
from pySim.esim.saip import param_source from pySim.esim.saip import param_source
from pySim.esim.saip import ProfileElement, ProfileElementSD, ProfileElementSequence from pySim.esim.saip import ProfileElement, ProfileElementSD, ProfileElementSequence
from pySim.esim.saip import ProfileElementHeader
from pySim.esim.saip import SecurityDomainKey, SecurityDomainKeyComponent from pySim.esim.saip import SecurityDomainKey, SecurityDomainKeyComponent
from pySim.global_platform import KeyUsageQualifier, KeyType 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: def unrpad(s: hexstr, c='f') -> hexstr:
return hexstr(s.rstrip(c)) return hexstr(s.rstrip(c))
@@ -297,9 +288,7 @@ class ConfigurableParameter(abc.ABC, metaclass=ClassVarMeta):
May be overridden by subclasses. May be overridden by subclasses.
This default implementation returns the maximum allowed value length -- a good fit for most subclasses. This default implementation returns the maximum allowed value length -- a good fit for most subclasses.
''' '''
l = cls.get_len_range()[1] or 16 return cls.get_len_range()[1] or 16
l = min(10*80, l)
return l
@classmethod @classmethod
def is_super_of(cls, other_class): def is_super_of(cls, other_class):
@@ -429,71 +418,69 @@ class BinaryParam(ConfigurableParameter):
class EnumParam(ConfigurableParameter): class EnumParam(ConfigurableParameter):
"""ConfigurableParameter for named value enumerations. """ConfigurableParameter for named integer enumeration values.
Subclasses define an own value_map, and implement their own apply_val() and get_values_from_pes(). 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
value_map = { be inherited from another mixin."""
# For example:
#'Meaningful label for value 23': 0x23, class Values(enum.IntEnum):
# Where 0x23 is a valid value to use for apply_val(), of any valid type. pass # subclasses override this
}
_value_map_reverse = None
@classmethod @classmethod
def validate_val(cls, val): def validate_val(cls, val) -> int:
orig_val = val if isinstance(val, int):
enum_val = None try:
if isinstance(val, str): return int(cls.Values(val))
enum_name = val except ValueError:
enum_val = cls.map_name_to_val(enum_name) pass
elif isinstance(val, str):
member = cls.map_name_to_val(val, strict=False)
if member is not None:
return member
# if the str is not one of the known value_map.keys(), is it maybe one of value_map.keys()? valid = ', '.join(m.name for m in cls.Values)
if enum_val is None and val in cls.value_map.values(): raise ValueError(f"{cls.get_name()}: invalid argument: {val!r}. Valid arguments are: {valid}")
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 @classmethod
def map_name_to_val(cls, name:str, strict=True): def map_name_to_val(cls, name: str, strict=True) -> int:
val = cls.value_map.get(name) """Return the integer value for a given enum member name. Performs an exact match first,
if val is not None: then falls back to fuzzy matching (case-insensitive, punctuation-insensitive)."""
return val try:
return int(cls.Values[name])
except KeyError:
pass
clean_name = cls.clean_name_str(name) clean = cls.clean_name_str(name)
for k, v in cls.value_map.items(): for member in cls.Values:
if clean_name == cls.clean_name_str(k): if cls.clean_name_str(member.name) == clean:
return v return int(member)
if strict: if strict:
raise ValueError(f"Problem in {cls.get_name()}: {name!r} is not a known value." valid = ', '.join(m.name for m in cls.Values)
f" Known values are: {cls.value_map.keys()!r}") raise ValueError(f"{cls.get_name()}: {name!r} is not a known value. Known values are: {valid}")
return None return None
@classmethod @classmethod
def map_val_to_name(cls, val, strict=False) -> str: def map_val_to_name(cls, val, strict=False) -> str:
if cls._value_map_reverse is None: """Return the enum member name for a given integer value."""
cls._value_map_reverse = dict((v, k) for k, v in cls.value_map.items()) try:
return cls.Values(val).name
name = cls._value_map_reverse.get(val) except ValueError:
if name: if strict:
return name raise ValueError(f"{cls.get_name()}: {val!r} ({type(val).__name__}) is not a known value.")
if strict: return None
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 @classmethod
def name_normalize(cls, name:str) -> str: def name_normalize(cls, name: str) -> str:
return cls.map_val_to_name(cls.map_name_to_val(name)) """Map a (possibly fuzzy) name to its canonical enum member name."""
return cls.Values(cls.map_name_to_val(name)).name
@classmethod @classmethod
def clean_name_str(cls, val): def clean_name_str(cls, val: str) -> str:
return re.sub('[^0-9A-Za-z-_]', '', val).lower() """Strip punctuation and case for fuzzy name comparison.
Treats hyphens and underscores as equivalent (both removed)."""
return re.sub('[^0-9A-Za-z]', '', val).lower()
class Iccid(DecimalParam): class Iccid(DecimalParam):
@@ -646,14 +633,13 @@ class SmspTpScAddr(ConfigurableParameter):
# - To generate the right amount of fillFileContent, pass total_len=42 to encode_record_bin(). # - 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 # - To show the right size in the PES, set f_smsp.rec_len = 42
ef_smsp_dec['alpha_id'] = '' 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. # re-encode into the File body.
f_smsp.body = ef_smsp.encode_record_bin(ef_smsp_dec, 1, total_len=f_smsp.rec_len) #
#print("SMSP (new): %s" % f_smsp.body)
# re-generate the pe.decoded member from the File instance # 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)
pe.file2pe(f_smsp) pe.file2pe(f_smsp)
@classmethod @classmethod
@@ -674,60 +660,6 @@ class SmspTpScAddr(ConfigurableParameter):
yield { cls.name: cls.tuple_to_str((international, digits)) } yield { cls.name: cls.tuple_to_str((international, digits)) }
class MncLen(EnumParam):
"""MNC length. Must be either 2 or 3. Sets only the MNC length field in EF-AD (Administrative Data)."""
name = 'MNC-LEN'
value_map = { '2': 2, '3': 3 }
default_source = param_source.ConstantSource
example_input = '2'
@classmethod
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'):
if not hasattr(pe, 'files'):
continue
f_ad = pe.files.get('ef-ad')
if not f_ad:
continue
# decode existing values
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)
pe.file2pe(f_ad)
@classmethod
def get_values_from_pes(cls, pes: ProfileElementSequence):
for pe in pes.get_pes_for_type('usim'):
if not hasattr(pe, 'files'):
continue
f_ad = pe.files.get('ef-ad', None)
if f_ad is None:
continue
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', None)
if mnc_len is None:
continue
yield { cls.name: cls.map_val_to_name(int(mnc_len)) }
class SdKey(BinaryParam): class SdKey(BinaryParam):
"""Configurable Security Domain (SD) Key. Value is presented as bytes. """Configurable Security Domain (SD) Key. Value is presented as bytes.
Non-abstract implementations are generated in SdKey.generate_sd_key_classes""" Non-abstract implementations are generated in SdKey.generate_sd_key_classes"""
@@ -1099,17 +1031,17 @@ class AlgorithmID(EnumParam, AlgoConfig):
"""use validate_val() from EnumParam, and apply_val() from AlgoConfig. """use validate_val() from EnumParam, and apply_val() from AlgoConfig.
In get_values_from_pes(), return enum value names, not raw values.""" In get_values_from_pes(), return enum value names, not raw values."""
name = "Algorithm" 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' algo_config_key = 'algorithmID'
example_input = "Milenage" example_input = "Milenage"
default_source = param_source.ConstantSource default_source = param_source.ConstantSource
# EnumParam.validate_val() returns the int values from value_map # 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
@classmethod @classmethod
def get_values_from_pes(cls, pes: ProfileElementSequence): def get_values_from_pes(cls, pes: ProfileElementSequence):
@@ -1156,7 +1088,7 @@ class MilenageRotationConstants(BinaryParam, AlgoConfig):
class MilenageXoringConstants(BinaryParam, AlgoConfig): class MilenageXoringConstants(BinaryParam, AlgoConfig):
"""XOR-ing constants c1,c2,c3,c4,c5 of Milenage, 128bit each. See 3GPP TS 35.206 Sections 2.3 + 5.3. """XOR-ing constants c1,c2,c3,c4,c5 of Milenage, 128bit each. See 3GPP TS 35.206 Sections 2.3 + 5.3.
Provided as octet-string concatenation of all 5 constants. The default value by 3GPP is the concatenation Provided as octet-string concatenation of all 5 constants. The default value by 3GPP is the concetenation
of:: of::
00000000000000000000000000000000 00000000000000000000000000000000
@@ -1184,411 +1116,3 @@ class TuakNumberOfKeccak(IntegerParam, AlgoConfig):
max_val = 255 max_val = 255
example_input = '1' example_input = '1'
default_source = param_source.ConstantSource 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'
+12
View File
@@ -863,6 +863,8 @@ class TransparentEF(CardEF):
t = self._tlv() if inspect.isclass(self._tlv) else self._tlv t = self._tlv() if inspect.isclass(self._tlv) else self._tlv
t.from_dict(abstract_data) t.from_dict(abstract_data)
return t.to_tlv() return t.to_tlv()
if 'raw' in abstract_data:
return h2b(abstract_data['raw'])
raise NotImplementedError( raise NotImplementedError(
"%s encoder not yet implemented. Patches welcome." % self) "%s encoder not yet implemented. Patches welcome." % self)
@@ -892,6 +894,8 @@ class TransparentEF(CardEF):
t = self._tlv() if inspect.isclass(self._tlv) else self._tlv t = self._tlv() if inspect.isclass(self._tlv) else self._tlv
t.from_dict(abstract_data) t.from_dict(abstract_data)
return b2h(t.to_tlv()) return b2h(t.to_tlv())
if 'raw' in abstract_data:
return abstract_data['raw']
raise NotImplementedError( raise NotImplementedError(
"%s encoder not yet implemented. Patches welcome." % self) "%s encoder not yet implemented. Patches welcome." % self)
@@ -1166,6 +1170,8 @@ class LinFixedEF(CardEF):
t = self._tlv() if inspect.isclass(self._tlv) else self._tlv t = self._tlv() if inspect.isclass(self._tlv) else self._tlv
t.from_dict(abstract_data) t.from_dict(abstract_data)
return b2h(t.to_tlv()) return b2h(t.to_tlv())
if 'raw' in abstract_data:
return abstract_data['raw']
raise NotImplementedError( raise NotImplementedError(
"%s encoder not yet implemented. Patches welcome." % self) "%s encoder not yet implemented. Patches welcome." % self)
@@ -1195,6 +1201,8 @@ class LinFixedEF(CardEF):
t = self._tlv() if inspect.isclass(self._tlv) else self._tlv t = self._tlv() if inspect.isclass(self._tlv) else self._tlv
t.from_dict(abstract_data) t.from_dict(abstract_data)
return t.to_tlv() return t.to_tlv()
if 'raw' in abstract_data:
return h2b(abstract_data['raw'])
raise NotImplementedError( raise NotImplementedError(
"%s encoder not yet implemented. Patches welcome." % self) "%s encoder not yet implemented. Patches welcome." % self)
@@ -1386,6 +1394,8 @@ class TransRecEF(TransparentEF):
t = self._tlv() if inspect.isclass(self._tlv) else self._tlv t = self._tlv() if inspect.isclass(self._tlv) else self._tlv
t.from_dict(abstract_data) t.from_dict(abstract_data)
return b2h(t.to_tlv()) return b2h(t.to_tlv())
if 'raw' in abstract_data:
return abstract_data['raw']
raise NotImplementedError( raise NotImplementedError(
"%s encoder not yet implemented. Patches welcome." % self) "%s encoder not yet implemented. Patches welcome." % self)
@@ -1415,6 +1425,8 @@ class TransRecEF(TransparentEF):
t = self._tlv() if inspect.isclass(self._tlv) else self._tlv t = self._tlv() if inspect.isclass(self._tlv) else self._tlv
t.from_dict(abstract_data) t.from_dict(abstract_data)
return t.to_tlv() return t.to_tlv()
if 'raw' in abstract_data:
return h2b(abstract_data['raw'])
raise NotImplementedError( raise NotImplementedError(
"%s encoder not yet implemented. Patches welcome." % self) "%s encoder not yet implemented. Patches welcome." % self)
+1 -1
View File
@@ -327,7 +327,7 @@ class EF_SUCI_Calc_Info(TransparentEF):
"""conversion method to generate list of {hnet_pubkey_identifier, hnet_pubkey} dicts """conversion method to generate list of {hnet_pubkey_identifier, hnet_pubkey} dicts
from flat [{hnet_pubkey_identifier: }, {net_pubkey: }, ...] list""" from flat [{hnet_pubkey_identifier: }, {net_pubkey: }, ...] list"""
out = [] out = []
while l: while len(l):
a = l.pop(0) a = l.pop(0)
b = l.pop(0) b = l.pop(0)
z = {**a, **b} z = {**a, **b}
Binary file not shown.
+25 -140
View File
@@ -21,7 +21,6 @@ import enum
import io import io
import sys import sys
import unittest import unittest
import json
from importlib import resources from importlib import resources
from osmocom.utils import hexstr from osmocom.utils import hexstr
from pySim.esim.saip import ProfileElementSequence from pySim.esim.saip import ProfileElementSequence
@@ -54,21 +53,18 @@ class ConfigurableParameterTest(unittest.TestCase):
def test_parameters(self): def test_parameters(self):
upp_fnames = ( upp_fnames = (
'SAIP2.1_gfmsuci.der', 'TS48v5_SAIP2.1A_NoBERTLV.der',
'TS48v5_SAIP2.3_BERTLV_SUCI.der',
'TS48v5_SAIP2.1B_NoBERTLV.der', 'TS48v5_SAIP2.1B_NoBERTLV.der',
'TS48v5_SAIP2.3_NoBERTLV.der', 'TS48v5_SAIP2.3_NoBERTLV.der',
) )
class Paramtest: class Paramtest:
iff_present_default = False def __init__(self, param_cls, val, expect_val, expect_clean_val=None):
def __init__(self, param_cls, val, expect_val, expect_clean_val=None, iff_present=None):
self.param_cls = param_cls self.param_cls = param_cls
self.val = val self.val = val
self.expect_clean_val = expect_clean_val self.expect_clean_val = expect_clean_val
self.expect_val = expect_val self.expect_val = expect_val
if iff_present is None:
iff_present = Paramtest.iff_present_default
self.iff_present = iff_present
param_tests = [ param_tests = [
Paramtest(param_cls=p13n.Imsi, val='123456', Paramtest(param_cls=p13n.Imsi, val='123456',
@@ -154,7 +150,7 @@ class ConfigurableParameterTest(unittest.TestCase):
Paramtest(param_cls=p13n.AlgorithmID, Paramtest(param_cls=p13n.AlgorithmID,
val='usim-test', val='usim-test',
expect_clean_val=3, expect_clean_val=3,
expect_val='usim-test'), expect_val='usim_test'),
Paramtest(param_cls=p13n.AlgorithmID, Paramtest(param_cls=p13n.AlgorithmID,
val=1, val=1,
@@ -167,7 +163,7 @@ class ConfigurableParameterTest(unittest.TestCase):
Paramtest(param_cls=p13n.AlgorithmID, Paramtest(param_cls=p13n.AlgorithmID,
val=3, val=3,
expect_clean_val=3, expect_clean_val=3,
expect_val='usim-test'), expect_val='usim_test'),
Paramtest(param_cls=p13n.K, Paramtest(param_cls=p13n.K,
val='01020304050607080910111213141516', val='01020304050607080910111213141516',
@@ -271,111 +267,7 @@ class ConfigurableParameterTest(unittest.TestCase):
'11111111111111111111111111111111' '11111111111111111111111111111111'
'22222222222222222222222222222222'), '22222222222222222222222222222222'),
Paramtest(param_cls=p13n.MncLen, ]
val='2',
expect_clean_val=2,
expect_val='2'),
Paramtest(param_cls=p13n.MncLen,
val=3,
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 ( for sdkey_cls in (
# thin out the number of tests, as a compromise between completeness and test runtime # thin out the number of tests, as a compromise between completeness and test runtime
@@ -468,8 +360,7 @@ class ConfigurableParameterTest(unittest.TestCase):
for t in param_tests: for t in param_tests:
test_idx += 1 test_idx += 1
testlog = [] logloc = f'{upp_fname} {t.param_cls.__name__}(val={valtypestr(t.val)})'
testlog.append(f'{upp_fname} {t.param_cls.__name__}(val={valtypestr(t.val)})')
param = None param = None
try: try:
@@ -477,32 +368,21 @@ class ConfigurableParameterTest(unittest.TestCase):
param.input_value = t.val param.input_value = t.val
param.validate() param.validate()
except ValueError as e: except ValueError as e:
raise ValueError(f'{" ".join(testlog)}: {e}') from e raise ValueError(f'{logloc}: {e}') from e
clean_val = param.value clean_val = param.value
testlog.append(f'clean_val={valtypestr(clean_val)}') logloc = f'{logloc} clean_val={valtypestr(clean_val)}'
if t.expect_clean_val is not None and t.expect_clean_val != clean_val: if t.expect_clean_val is not None and t.expect_clean_val != clean_val:
raise ValueError(f'{" ".join(testlog)}: expected' raise ValueError(f'{logloc}: expected'
f' expect_clean_val={valtypestr(t.expect_clean_val)}') f' expect_clean_val={valtypestr(t.expect_clean_val)}')
# on my laptop, deepcopy is about 30% slower than decoding the DER from scratch: # on my laptop, deepcopy is about 30% slower than decoding the DER from scratch:
# pes = copy.deepcopy(orig_pes) # pes = copy.deepcopy(orig_pes)
pes = ProfileElementSequence.from_der(der) 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: try:
param.apply(pes) param.apply(pes)
except ValueError as e: except ValueError as e:
raise ValueError(f'{" ".join(testlog)} apply_val(clean_val): {e}') from e raise ValueError(f'{logloc} apply_val(clean_val): {e}') from e
changed_der = pes.to_der() changed_der = pes.to_der()
@@ -520,18 +400,22 @@ class ConfigurableParameterTest(unittest.TestCase):
else: else:
read_back_val_type = f'{type(read_back_val).__name__}' read_back_val_type = f'{type(read_back_val).__name__}'
testlog.append(f'read_back_val={valtypestr(read_back_val)}') logloc = (f'{logloc} 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(): if isinstance(read_back_val, dict) and not t.param_cls.get_name() in read_back_val.keys():
raise ValueError(f'{" ".join(testlog)}: expected to find name {t.param_cls.get_name()!r} in read_back_val') raise ValueError(f'{logloc}: expected to find name {t.param_cls.get_name()!r} in read_back_val')
expect_val = t.expect_val expect_val = t.expect_val
if not isinstance(expect_val, dict): if not isinstance(expect_val, dict):
expect_val = { t.param_cls.get_name(): expect_val } expect_val = { t.param_cls.get_name(): expect_val }
if read_back_val != expect_val: if read_back_val != expect_val:
raise ValueError(f'{" ".join(testlog)}: expected {expect_val=!r}:{type(t.expect_val).__name__}') raise ValueError(f'{logloc}: expected {expect_val=!r}:{type(t.expect_val).__name__}')
output = "\nok: " + "\n ".join(testlog) ok = logloc.replace(' clean_val', '\n\tclean_val'
).replace(' read_back_val', '\n\tread_back_val'
).replace('=', '=\t'
)
output = f'\nok: {ok}'
outputs.append(output) outputs.append(output)
print(output) print(output)
@@ -667,7 +551,7 @@ class TestEnumParam(unittest.TestCase):
def test_validate_by_name_exact(self): def test_validate_by_name_exact(self):
self.assertEqual(p13n.AlgorithmID.validate_val('Milenage'), 1) self.assertEqual(p13n.AlgorithmID.validate_val('Milenage'), 1)
self.assertEqual(p13n.AlgorithmID.validate_val('TUAK'), 2) 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): def test_validate_by_int(self):
self.assertEqual(p13n.AlgorithmID.validate_val(1), 1) self.assertEqual(p13n.AlgorithmID.validate_val(1), 1)
@@ -680,6 +564,7 @@ class TestEnumParam(unittest.TestCase):
self.assertEqual(p13n.AlgorithmID.validate_val('tuak'), 2) self.assertEqual(p13n.AlgorithmID.validate_val('tuak'), 2)
def test_validate_fuzzy_hyphen_underscore(self): 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) self.assertEqual(p13n.AlgorithmID.validate_val('usim-test'), 3)
def test_validate_invalid_name(self): def test_validate_invalid_name(self):
@@ -716,7 +601,7 @@ class TestEnumParam(unittest.TestCase):
def test_map_val_known(self): def test_map_val_known(self):
self.assertEqual(p13n.AlgorithmID.map_val_to_name(1), 'Milenage') 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(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): def test_map_val_unknown_nonstrict(self):
self.assertIsNone(p13n.AlgorithmID.map_val_to_name(99)) self.assertIsNone(p13n.AlgorithmID.map_val_to_name(99))
@@ -730,13 +615,13 @@ class TestEnumParam(unittest.TestCase):
def test_name_normalize(self): 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('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('usim-test'), 'usim_test')
# --- clean_name_str --- # --- clean_name_str ---
def test_clean_name_str(self): def test_clean_name_str(self):
self.assertEqual(p13n.AlgorithmID.clean_name_str('usim-test'), 'usim-test') self.assertEqual(p13n.AlgorithmID.clean_name_str('usim-test'), 'usimtest')
self.assertEqual(p13n.AlgorithmID.clean_name_str('usim_test'), 'usim_test') self.assertEqual(p13n.AlgorithmID.clean_name_str('usim_test'), 'usimtest')
self.assertEqual(p13n.AlgorithmID.clean_name_str('Milenage'), 'milenage') self.assertEqual(p13n.AlgorithmID.clean_name_str('Milenage'), 'milenage')
self.assertEqual(p13n.AlgorithmID.clean_name_str('foo bar!'), 'foobar') self.assertEqual(p13n.AlgorithmID.clean_name_str('foo bar!'), 'foobar')
File diff suppressed because it is too large Load Diff