laforge has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/29033 ) Change subject: Add new pySim.ota library, implement SIM OTA crypto ...................................................................... Add new pySim.ota library, implement SIM OTA crypto This introduces a hierarchy of classes implementing * ETS TS 102 225 (general command structure) * 3GPP TS 31.115 (dialects for SMS-PP) In this initial patch only the SMS "dialect" is supported, but it is foreseen that USSD/SMSCB/HTTPS dialects can be added at a later point. Change-Id: I193ff4712c8503279c017b4b1324f0c3d38b9f84 --- A pySim/ota.py A tests/test_ota.py 2 files changed, 536 insertions(+), 0 deletions(-) git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/33/29033/1 diff --git a/pySim/ota.py b/pySim/ota.py new file mode 100644 index 0000000..4cac28d --- /dev/null +++ b/pySim/ota.py @@ -0,0 +1,451 @@ +"""Code related to SIM/UICC OTA according to TS 102 225 + TS 31.115.""" + +# (C) 2021-2022 by Harald Welte &lt;laforge(a)osmocom.org&gt; +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 2 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 General Public License for more details. +# +# 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 pySim.construct import * +from pySim.utils import b2h +from pySim.sms import UserDataHeader +from construct import * +from bidict import bidict +import zlib +import abc +import struct + +# ETS TS 102 225 gives the general command structure and the dialects for CAT_TP, TCP/IP and HTTPS +# 3GPP TS 31.115 gives the dialects for SMS-PP, SMS-CB, USSD and HTTP + +# CPI CPL CHI CHL SPI KIc KID TAR CNTR PCNTR RC/CC/DS data + +# CAT_TP TCP/IP SMS +# CPI 0x01 0x01 =IEIa=70,len=0 +# CHI NULL NULL NULL +# CPI, CPL and CHL included in RC/CC/DS true true +# RPI 0x02 0x02 =IEIa=71,len=0 +# RHI NULL NULL +# RPI, RPL and RHL included in RC/CC/DS true true +# packet-id 0-bf,ff 0-bf,ff +# identification packet false 102 225 tbl 6 + +# KVN 1..f; KI1=KIc, KI2=KID, KI3=DEK + +# TS 102 225 Table 5 +ota_status_codes = bidict({ + 0x00: 'PoR OK', + 0x01: 'RC/CC/DS failed', + 0x02: 'CNTR low', + 0x03: 'CNTR high', + 0x04: 'CNTR blocked', + 0x05: 'Ciphering error', + 0x06: 'Unidentified security error', + 0x07: 'Insufficient memory', + 0x08: 'more time', + 0x09: 'TAR unknown', + 0x0a: 'Insufficient security level', + 0x0b: 'Actual Response in SMS-SUBMIT', # 31.115 + 0x0c: 'Actual Response in USSD', # 31.115 +}) + +# ETSI TS 102 225 Table 5 + 3GPP TS 31.115 Section 7 +ResponseStatus = Enum(Int8ub, por_ok=0, rc_cc_ds_failed=1, cntr_low=2, cntr_high=3, + cntr_blocked=4, ciphering_error=5, undefined_security_error=6, + insufficient_memory=7, more_time_needed=8, tar_unknown=9, + insufficient_security_level=0x0A, + actual_response_sms_submit=0x0B, + actual_response_ussd=0x0C) + +# ETSI TS 102 226 Section 5.1.2 +CompactRemoteResp = Struct('number_of_commands'/Int8ub, + 'last_status_word'/HexAdapter(Bytes(2)), + 'last_response_data'/HexAdapter(GreedyBytes)) + +RC_CC_DS = Enum(BitsInteger(2), no_rc_cc_ds=0, rc=1, cc=2, ds=3) + +# TS 102 225 Section 5.1.1 + TS 31.115 Section 4.2 +SPI = BitStruct( # first octet + Padding(3), + 'counter'/Enum(BitsInteger(2), no_counter=0, counter_no_replay_or_seq=1, + counter_must_be_higher=2, counter_must_be_lower=3), + 'ciphering'/Flag, + 'rc_cc_ds'/RC_CC_DS, + # second octet + Padding(2), + 'por_in_submit'/Flag, + 'por_shall_be_ciphered'/Flag, + 'por_rc_cc_ds'/RC_CC_DS, + 'por'/Enum(BitsInteger(2), no_por=0, + por_required=1, por_only_when_error=2) +) + +# TS 102 225 Section 5.1.2 +KIC = BitStruct('key'/BitsInteger(4), + 'algo'/Enum(BitsInteger(4), implicit=0, single_des=1, triple_des_cbc2=5, triple_des_cbc3=9, + aes_cbc=2) + ) + +# TS 102 225 Section 5.1.3.1 +KID_CC = BitStruct('key'/BitsInteger(4), + 'algo'/Enum(BitsInteger(4), implicit=0, single_des=1, triple_des_cbc2=5, triple_des_cbc3=9, + aes_cmac=2) + ) + +# TS 102 225 Section 5.1.3.2 +KID_RC = BitStruct('key'/BitsInteger(4), + 'algo'/Enum(BitsInteger(4), implicit=0, crc16=1, crc32=5, proprietary=3) + ) + +SmsCommandPacket = Struct('cmd_pkt_len'/Int16ub, + 'cmd_hdr_len'/Int8ub, + 'spi'/SPI, + 'kic'/KIC, + 'kid'/Switch(this.spi.rc_cc_ds, {'cc': KID_CC, 'rc': KID_RC }), + 'tar'/Bytes(3), + 'secured_data'/GreedyBytes) + +class OtaKeyset: + """The OTA related data (key material, counter) to be used in encrypt/decrypt.""" + def __init__(self, algo_crypt: str, kic_idx: int, kic: bytes, + algo_auth: str, kid_idx: int, kid: bytes, cntr: int = 0): + self.algo_crypt = algo_crypt + self.kic = bytes(kic) + self.kic_idx = kic_idx + self.algo_auth = algo_auth + self.kid = bytes(kid) + self.kid_idx = kid_idx + self.cntr = cntr + + @property + def auth(self): + """Return an instance of the matching OtaAlgoAuth.""" + return OtaAlgoAuth.fromKeyset(self) + + @property + def crypt(self): + """Return an instance of the matching OtaAlgoCrypt.""" + return OtaAlgoCrypt.fromKeyset(self) + +class OtaCheckError(Exception): + pass + +class OtaDialect(abc.ABC): + """Base Class for OTA dialects such as SMS, BIP, ...""" + + def _compute_sig_len(self, spi:SPI): + if spi['rc_cc_ds'] == 'no_rc_cc_ds': + return 0 + elif spi['rc_cc_ds'] == 'rc': # CRC-32 + return 4 + elif spi['rc_cc_ds'] == 'cc': # Cryptographic Checksum (CC) + # TODO: this is not entirely correct, as in AES case it could be 4 or 8 + return 8 + else: + raise ValueError("Invalid rc_cc_ds: %s" % spi['rc_cc_ds']) + + @abc.abstractmethod + def encode_cmd(self, otak: OtaKeyset, tar: bytes, apdu: bytes) -> bytes: + pass + + @abc.abstractmethod + def decode_resp(self, otak: OtaKeyset, apdu: bytes) -> bytes: + pass + + +from Cryptodome.Cipher import DES, DES3, AES +from Cryptodome.Hash import CMAC + +class OtaAlgo(abc.ABC): + iv = b'\x00\x00\x00\x00\x00\x00\x00\x00' + blocksize = None + + @staticmethod + def _get_padding(in_len: int, multiple: int, padding: int = 0): + """Return padding bytes towards multiple of N.""" + if in_len % multiple == 0: + return b'' + pad_cnt = multiple - (in_len % multiple) + return b'\x00' * pad_cnt + + @staticmethod + def _pad_to_multiple(indat: bytes, multiple: int, padding: int = 0): + """Pad input bytes to multiple of N.""" + return indat + OtaAlgo._get_padding(len(indat), multiple, padding) + + def pad_to_blocksize(self, indat: bytes, padding: int = 0): + """Pad the given input data to multiple of the cipher block size.""" + return self._pad_to_multiple(indat, self.blocksize, padding) + + def __init__(self, otak: OtaKeyset): + self.otak = otak + + def __str__(self): + return self.__class__.__name__ + +class OtaAlgoCrypt(OtaAlgo, abc.ABC): + def __init__(self, otak: OtaKeyset): + if self.enum_name != otak.algo_crypt: + raise ValueError('Cannot use algorithm %s with key for %s' % (self.enum_name, otak.algo_crypt)) + super().__init__(otak) + + def encrypt(self, data:bytes) -> bytes: + """Encrypt given input bytes using the key material given in constructor.""" + padded_data = self.pad_to_blocksize(data) + return self._encrypt(data) + + def decrypt(self, data:bytes) -> bytes: + """Decrypt given input bytes using the key material given in constructor.""" + return self._decrypt(data) + + @abc.abstractmethod + def _encrypt(self, data:bytes) -> bytes: + """Actual implementation, to be implemented by derived class.""" + pass + + @abc.abstractmethod + def _decrypt(self, data:bytes) -> bytes: + """Actual implementation, to be implemented by derived class.""" + pass + + @classmethod + def fromKeyset(cls, otak: OtaKeyset) -> 'OtaAlgoCrypt': + """Resolve the class for the encryption algorithm of otak and instantiate it.""" + for subc in cls.__subclasses__(): + if subc.enum_name == otak.algo_crypt: + return subc(otak) + raise ValueError('No implementation for crypt algorithm %s' % otak.algo_auth) + +class OtaAlgoAuth(OtaAlgo, abc.ABC): + def __init__(self, otak: OtaKeyset): + if self.enum_name != otak.algo_auth: + raise ValueError('Cannot use algorithm %s with key for %s' % (self.enum_name, otak.algo_crypt)) + super().__init__(otak) + + def sign(self, data:bytes) -> bytes: + """Compute the CC/CR check bytes for the input data using key material + given in constructor.""" + padded_data = self.pad_to_blocksize(data) + sig = self._sign(padded_data) + return sig + + def check_sig(self, data:bytes, cc_received:bytes): + """Compute the CC/CR check bytes for the input data and compare against cc_received.""" + cc = self.sign(data) + if cc_received != cc: + raise OtaCheckError('Received CC (%s) != Computed CC (%s)' % (b2h(cc_received), b2h(cc))) + + @abc.abstractmethod + def _sign(self, data:bytes) -> bytes: + """Actual implementation, to be implemented by derived class.""" + pass + + @classmethod + def fromKeyset(cls, otak: OtaKeyset) -> 'OtaAlgoAuth': + """Resolve the class for the authentication algorithm of otak and instantiate it.""" + for subc in cls.__subclasses__(): + if subc.enum_name == otak.algo_auth: + return subc(otak) + raise ValueError('No implementation for auth algorithm %s' % otak.algo_auth) + +class OtaAlgoCryptDES(OtaAlgoCrypt): + """DES is insecure. For backwards compatibility with pre-Rel8""" + name = 'DES' + enum_name = 'single_des' + blocksize = 8 + def _encrypt(self, data:bytes) -> bytes: + cipher = DES.new(self.otak.kic, DES.MODE_CBC, self.iv) + return cipher.encrypt(data) + + def _decrypt(self, data:bytes) -> bytes: + cipher = DES.new(self.otak.kic, DES.MODE_CBC, self.iv) + return cipher.decrypt(data) + +class OtaAlgoAuthDES(OtaAlgoAuth): + """DES is insecure. For backwards compatibility with pre-Rel8""" + name = 'DES' + enum_name = 'single_des' + blocksize = 8 + def _sign(self, data:bytes) -> bytes: + cipher = DES.new(self.otak.kid, DES.MODE_CBC, self.iv) + ciph = cipher.encrypt(data) + return ciph[len(ciph) - 8:] + +class OtaAlgoCryptDES3(OtaAlgoCrypt): + name = '3DES' + enum_name = 'triple_des_cbc2' + blocksize = 8 + def _encrypt(self, data:bytes) -> bytes: + cipher = DES3.new(self.otak.kic, DES3.MODE_CBC, self.iv) + return cipher.encrypt(data) + + def _decrypt(self, data:bytes) -> bytes: + cipher = DES3.new(self.otak.kic, DES3.MODE_CBC, self.iv) + return cipher.decrypt(data) + +class OtaAlgoAuthDES3(OtaAlgoAuth): + name = '3DES' + enum_name = 'triple_des_cbc2' + blocksize = 8 + def _sign(self, data:bytes) -> bytes: + cipher = DES3.new(self.otak.kid, DES3.MODE_CBC, self.iv) + ciph = cipher.encrypt(data) + return ciph[len(ciph) - 8:] + +class OtaAlgoCryptAES(OtaAlgoCrypt): + name = 'AES' + enum_name = 'aes_cbc' + blocksize = 16 # TODO: is this needed? + def _encrypt(self, data:bytes) -> bytes: + cipher = AES.new(self.otak.kic, AES.MODE_CBC, self.iv) + return cipher.encrypt(data) + + def _decrypt(self, data:bytes) -> bytes: + cipher = AES.new(self.otak.kic, AES.MODE_CBC, self.iv) + return cipher.decrypt(data) + +class OtaAlgoAuthAES(OtaAlgoAuth): + name = 'AES' + enum_name = 'aes_cmac' + blocksize = 16 # TODO: is this needed? + def _sign(self, data:bytes) -> bytes: + cmac = CMAC.new(self.otak.kid, ciphermod=AES, mac_len=8) + cmac.update(data) + ciph = cmac.digest() + return ciph[len(ciph) - 8:] + + + +class OtaDialectSms(OtaDialect): + """OTA dialect for SMS based transport, as described in 3GPP TS 31.115.""" + SmsResponsePacket = Struct('rpl'/Int16ub, + 'rhl'/Int8ub, + 'tar'/Bytes(3), + 'cntr'/Bytes(5), + 'pcntr'/Int8ub, + 'response_status'/ResponseStatus, + 'cc_rc'/Bytes(this.rhl-10), + 'secured_data'/GreedyBytes) + + def encode_cmd(self, otak: OtaKeyset, tar: bytes, spi: dict, apdu: bytes) -> bytes: + # length of signature in octets + len_sig = self._compute_sig_len(spi) + pad_cnt = 0 + if spi['ciphering']: # ciphering is requested + # append padding bytes to end up with blocksize + len_cipher = 6 + len_sig + len(apdu) + apdu += otak.crypt._get_padding(len_cipher, otak.crypt.blocksize) + + kic = {'key': otak.kic_idx, 'algo': otak.algo_crypt} + kid = {'key': otak.kid_idx, 'algo': otak.algo_auth} + + # CHL = number of octets from (and including) SPI to the end of RC/CC/DS + # 13 == SPI(2) + KIc(1) + KId(1) + TAR(3) + CNTR(5) + PCNTR(1) + chl = 13 + len_sig + + # CHL + SPI (+ KIC + KID) + c = Struct('chl'/Int8ub, 'spi'/SPI, 'kic'/KIC, 'kid'/KID_CC, 'tar'/Bytes(3)) + part_head = c.build({'chl': chl, 'spi':spi, 'kic':kic, 'kid':kid, 'tar':tar}) + #print("part_head: %s" % b2h(part_head)) + + # CNTR + PCNTR (CNTR not used) + part_cnt = otak.cntr.to_bytes(5, 'big') + pad_cnt.to_bytes(1, 'big') + #print("part_cnt: %s" % b2h(part_cnt)) + + envelope_data = part_head + part_cnt + apdu + #print("envelope_data: %s" % b2h(envelope_data)) + + # 2-byte CPL. CPL is part of RC/CC/CPI to end of secured data, including any padding for ciphering + # CPL from and including CPI to end of secured data, including any padding for ciphering + cpl = len(envelope_data) + len_sig + envelope_data = cpl.to_bytes(2, 'big') + envelope_data + #print("envelope_data with cpl: %s" % b2h(envelope_data)) + + if spi['rc_cc_ds'] == 'cc': + cc = otak.auth.sign(envelope_data) + envelope_data = part_cnt + cc + apdu + elif spi['rc_cc_ds'] == 'rc': + # CRC32 + crc32 = zlib.crc32(envelope_data) & 0xffffffff + envelope_data = part_cnt + crc32.to_bytes(4, 'big') + apdu + elif spi['rc_cc_ds'] == 'no_rc_cc_ds': + envelope_data = part_cnt + apdu + else: + raise ValueError("Invalid rc_cc_ds: %s" % spi['rc_cc_ds']) + + #print("envelope_data with sig: %s" % b2h(envelope_data)) + + # encrypt as needed + if spi['ciphering']: # ciphering is requested + ciph = otak.crypt.encrypt(envelope_data) + envelope_data = part_head + ciph + # prefix with another CPL + cpl = len(envelope_data) + envelope_data = cpl.to_bytes(2, 'big') + envelope_data + else: + envelope_data = part_head + envelope_data + + #print("envelope_data: %s" % b2h(envelope_data)) + + return envelope_data + + def decode_resp(self, otak: OtaKeyset, spi: dict, data: bytes) -> bytes: + if isinstance(data, str): + data = h2b(data) + # plain-text POR: 027100000e0ab000110000000000000001612f + # UDHL RPI IEDLa RPL RHL TAR CNTR PCNTR STS + # 02 71 00 000e 0a b00011 0000000000 00 00 01 612f + # POR with CC: 027100001612b000110000000000000055f47118381175fb01612f + # POR with CC+CIPH: 027100001c12b000119660ebdb81be189b5e4389e9e7ab2bc0954f963ad869ed7c + if data[0] != 0x02: + raise ValueError('Unexpected UDL=0x%02x' % data[0]) + udhd, remainder = UserDataHeader.fromBytes(data) + if not udhd.has_ie(0x71): + raise ValueError('RPI 0x71 not found in UDH') + rph_rhl_tar = remainder[:6] # RPH+RHL+TAR; not ciphered + res = self.SmsResponsePacket.parse(remainder) + + if spi['por_shall_be_ciphered']: + # decrypt + ciphered_part = remainder[6:] + deciph = otak.crypt.decrypt(ciphered_part) + temp_data = rph_rhl_tar + deciph + res = self.SmsResponsePacket.parse(temp_data) + # remove specified number of padding bytes, if any + if res['pcntr'] != 0: + # this conditional is needed as python [:-0] renders an empty return! + res['secured_data'] = res['secured_data'][:-res['pcntr']] + remainder = temp_data + + # is there a CC/RC present? + len_sig = res['rhl'] - 10 + if spi['por_rc_cc_ds'] == 'no_rc_cc_ds': + if len_sig: + raise OtaCheckError('No RC/CC/DS requested, but len_sig=%u' % len_sig) + elif spi['por_rc_cc_ds'] == 'cc': + # verify signature + # UDH is part of CC/RC! + udh = data[:3] + # RPL, RHL, TAR, CNTR, PCNTR and STSare part of CC/RC + rpl_rhl_tar_cntr_pcntr_sts = remainder[:13] + # remove the CC/RC bytes + temp_data = udh + rpl_rhl_tar_cntr_pcntr_sts + remainder[13+len_sig:] + cc = otak.auth.check_sig(temp_data, res['cc_rc']) + # TODO: CRC + else: + raise OtaCheckError('Unknown por_rc_cc_ds: %s' % spi['por_rc_cc_ds']) + + # TODO: ExpandedRemoteResponse according to TS 102 226 5.2.2 + dec = CompactRemoteResp.parse(res['secured_data']) + dec['tar'] = res['tar'] + dec['response_status'] = res['response_status'] + return dec diff --git a/tests/test_ota.py b/tests/test_ota.py new file mode 100644 index 0000000..e8f56f8 --- /dev/null +++ b/tests/test_ota.py @@ -0,0 +1,85 @@ +#!/usr/bin/env python3 + +import unittest +from pySim.utils import h2b, b2h +from pySim.ota import * + +class Test_SMS_3DES(unittest.TestCase): + tar = h2b('b00000') + """Test the OtaDialectSms for 3DES algorithms.""" + def __init__(self, foo, **kwargs): + super().__init__(foo, **kwargs) + # KIC1 + KID1 of 8988211000000467285 + KIC1 = h2b('D0FDA31990D8D64178601317191669B4') + KID1 = h2b('D24EB461799C5E035C77451FD9404463') + KIC3 = h2b('C21DD66ACAC13CB3BC8B331B24AFB57B') + KID3 = h2b('12110C78E678C25408233076AA033615') + self.od = OtaKeyset(algo_crypt='triple_des_cbc2', kic_idx=3, kic=KIC3, + algo_auth='triple_des_cbc2', kid_idx=3, kid=KID3) + self.dialect = OtaDialectSms() + self.spi_base = { + 'counter':'no_counter', + 'ciphering': True, + 'rc_cc_ds': 'cc', + 'por_in_submit':False, + 'por': 'por_required', + 'por_shall_be_ciphered': True, + 'por_rc_cc_ds': 'cc', + } + + def _check_response(self, r): + self.assertEqual(r['number_of_commands'], 1) + self.assertEqual(r['last_status_word'], '612f') + self.assertEqual(r['last_response_data'], u'') + self.assertEqual(r['response_status'], 'por_ok') + + def test_resp_3des_ciphered(self): + spi = self.spi_base + spi['por_shall_be_ciphered'] = True + spi['por_rc_cc_ds'] = 'cc' + r = self.dialect.decode_resp(self.od, spi, '027100001c12b000119660ebdb81be189b5e4389e9e7ab2bc0954f963ad869ed7c') + self._check_response(r) + + def test_resp_3des_signed(self): + spi = self.spi_base + spi['por_shall_be_ciphered'] = False + spi['por_rc_cc_ds'] = 'cc' + r = self.dialect.decode_resp(self.od, spi, '027100001612b000110000000000000055f47118381175fb01612f') + self._check_response(r) + + def test_resp_3des_signed_err(self): + """Expect an OtaCheckError exception if the computed CC != received CC""" + spi = self.spi_base + spi['por_shall_be_ciphered'] = False + spi['por_rc_cc_ds'] = 'cc' + with self.assertRaises(OtaCheckError) as context: + r = self.dialect.decode_resp(self.od, spi, '027100001612b000110000000000000055f47118381175fb02612f') + self.assertTrue('!= Computed CC' in str(context.exception)) + + def test_resp_3des_none(self): + spi = self.spi_base + spi['por_shall_be_ciphered'] = False + spi['por_rc_cc_ds'] = 'no_rc_cc_ds' + r = self.dialect.decode_resp(self.od, spi, '027100000e0ab000110000000000000001612f') + self._check_response(r) + + def test_cmd_3des_ciphered(self): + spi = self.spi_base + spi['ciphering'] = True + spi['rc_cc_ds'] = 'no_rc_cc_ds' + r = self.dialect.encode_cmd(self.od, self.tar, spi, h2b('00a40000023f00')) + self.assertEqual(b2h(r), '00180d04193535b000000c8478b552a4ffc5a8f099b83cad7123') + + def test_cmd_3des_signed(self): + spi = self.spi_base + spi['ciphering'] = False + spi['rc_cc_ds'] = 'cc' + r = self.dialect.encode_cmd(self.od, self.tar, spi, h2b('00a40000023f00')) + self.assertEqual(b2h(r), '1502193535b00000000000000000072ea17bdb72060e00a40000023f00') + + def test_cmd_3des_none(self): + spi = self.spi_base + spi['ciphering'] = False + spi['rc_cc_ds'] = 'no_rc_cc_ds' + r = self.dialect.encode_cmd(self.od, self.tar, spi, h2b('00a40000023f00')) + self.assertEqual(b2h(r), '0d00193535b0000000000000000000a40000023f00') -- To view, visit https://gerrit.osmocom.org/c/pysim/+/29033 To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings Gerrit-Project: pysim Gerrit-Branch: master Gerrit-Change-Id: I193ff4712c8503279c017b4b1324f0c3d38b9f84 Gerrit-Change-Number: 29033 Gerrit-PatchSet: 1 Gerrit-Owner: laforge &lt;laforge(a)osmocom.org&gt; Gerrit-MessageType: newchange