diff --git a/pySim/ota.py b/pySim/ota.py index 14bad565..10de6116 100644 --- a/pySim/ota.py +++ b/pySim/ota.py @@ -397,7 +397,7 @@ class OtaDialectSms(OtaDialect): # POR with CC+CIPH: 027100001c12b000119660ebdb81be189b5e4389e9e7ab2bc0954f963ad869ed7c if data[0] != 0x02: raise ValueError('Unexpected UDL=0x%02x' % data[0]) - udhd, remainder = UserDataHeader.fromBytes(data) + udhd, remainder = UserDataHeader.from_bytes(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 diff --git a/pySim/sms.py b/pySim/sms.py index 88ee27a1..5ba5ec8e 100644 --- a/pySim/sms.py +++ b/pySim/sms.py @@ -20,7 +20,7 @@ import typing import abc from bidict import bidict -from construct import Int8ub, Byte, Bytes, Bit, Flag, BitsInteger, Flag +from construct import Int8ub, Byte, Bytes, Bit, Flag, BitsInteger from construct import Struct, Enum, Tell, BitStruct, this, Padding from construct import Prefixed, GreedyRange, GreedyBytes @@ -51,13 +51,13 @@ class UserDataHeader: return False @classmethod - def fromBytes(cls, inb: BytesOrHex) -> typing.Tuple['UserDataHeader', bytes]: + def from_bytes(cls, inb: BytesOrHex) -> typing.Tuple['UserDataHeader', bytes]: if isinstance(inb, str): inb = h2b(inb) res = cls._construct.parse(inb) return cls(res['ies']), res['data'] - def toBytes(self) -> bytes: + def to_bytes(self) -> bytes: return self._construct.build({'ies':self.ies, 'data':b''}) @@ -117,7 +117,7 @@ class AddressField: return 'AddressField(TON=%s, NPI=%s, %s)' % (self.ton, self.npi, self.digits) @classmethod - def fromBytes(cls, inb: BytesOrHex) -> typing.Tuple['AddressField', bytes]: + def from_bytes(cls, inb: BytesOrHex) -> typing.Tuple['AddressField', bytes]: """Construct an AddressField instance from the binary T-PDU address format.""" if isinstance(inb, str): inb = h2b(inb) @@ -129,16 +129,16 @@ class AddressField: return cls(res['digits'][:res['addr_len']], ton, npi), inb[res['tell']:] @classmethod - def fromSmpp(cls, addr, ton, npi) -> 'AddressField': + def from_smpp(cls, addr, ton, npi) -> 'AddressField': """Construct an AddressField from {source,dest}_addr_{,ton,npi} attributes of smpp.pdu.""" # return the resulting instance return cls(addr.decode('ascii'), AddressField.smpp_map_ton[ton.name], AddressField.smpp_map_npi[npi.name]) - def toSmpp(self): + def to_smpp(self): """Return smpp.pdo.*.source,dest}_addr_{,ton,npi} attributes for given AddressField.""" return (self.digits, self.smpp_map_ton.inverse[self.ton], self.smpp_map_npi.inverse[self.npi]) - def toBytes(self) -> bytes: + def to_bytes(self) -> bytes: """Encode the AddressField into the binary representation as used in T-PDU.""" num_digits = len(self.digits) if num_digits % 2: @@ -185,13 +185,12 @@ class SMS_DELIVER(SMS_TPDU): return '%s(MTI=%s, MMS=%s, LP=%s, RP=%s, UDHI=%s, SRI=%s, OA=%s, PID=%2x, DCS=%x, SCTS=%s, UDL=%u, UD=%s)' % (self.__class__.__name__, self.tp_mti, self.tp_mms, self.tp_lp, self.tp_rp, self.tp_udhi, self.tp_sri, self.tp_oa, self.tp_pid, self.tp_dcs, self.tp_scts, self.tp_udl, self.tp_ud) @classmethod - def fromBytes(cls, inb: BytesOrHex) -> 'SMS_DELIVER': + def from_bytes(cls, inb: BytesOrHex) -> 'SMS_DELIVER': """Construct a SMS_DELIVER instance from the binary encoded format as used in T-PDU.""" if isinstance(inb, str): inb = h2b(inb) - flags = inb[0] d = SMS_DELIVER.flags_construct.parse(inb) - oa, remainder = AddressField.fromBytes(inb[1:]) + oa, remainder = AddressField.from_bytes(inb[1:]) d['tp_oa'] = oa offset = 0 d['tp_pid'] = remainder[offset] @@ -206,7 +205,7 @@ class SMS_DELIVER(SMS_TPDU): d['tp_ud'] = remainder[offset:] return cls(**d) - def toBytes(self) -> bytes: + def to_bytes(self) -> bytes: """Encode a SMS_DELIVER instance to the binary encoded format as used in T-PDU.""" outb = bytearray() d = { @@ -215,7 +214,7 @@ class SMS_DELIVER(SMS_TPDU): } flags = SMS_DELIVER.flags_construct.build(d) outb.extend(flags) - outb.extend(self.tp_oa.toBytes()) + outb.extend(self.tp_oa.to_bytes()) outb.append(self.tp_pid) outb.append(self.tp_dcs) outb.extend(self.tp_scts) @@ -225,18 +224,18 @@ class SMS_DELIVER(SMS_TPDU): return outb @classmethod - def fromSmpp(cls, smpp_pdu) -> 'SMS_DELIVER': + def from_smpp(cls, smpp_pdu) -> 'SMS_DELIVER': """Construct a SMS_DELIVER instance from the deliver format used by smpp.pdu.""" if smpp_pdu.id == pdu_types.CommandId.submit_sm: - return cls.fromSmppSubmit(smpp_pdu) + return cls.from_smpp_submit(smpp_pdu) else: raise ValueError('Unsupported SMPP commandId %s' % smpp_pdu.id) @classmethod - def fromSmppSubmit(cls, smpp_pdu) -> 'SMS_DELIVER': + def from_smpp_submit(cls, smpp_pdu) -> 'SMS_DELIVER': """Construct a SMS_DELIVER instance from the submit format used by smpp.pdu.""" ensure_smpp_is_8bit(smpp_pdu.params['data_coding']) - tp_oa = AddressField.fromSmpp(smpp_pdu.params['source_addr'], + tp_oa = AddressField.from_smpp(smpp_pdu.params['source_addr'], smpp_pdu.params['source_addr_ton'], smpp_pdu.params['source_addr_npi']) tp_ud = smpp_pdu.params['short_message'] @@ -276,7 +275,7 @@ class SMS_SUBMIT(SMS_TPDU): return '%s(MTI=%s, RD=%s, VPF=%u, RP=%s, UDHI=%s, SRR=%s, DA=%s, PID=%2x, DCS=%x, VP=%s, UDL=%u, UD=%s)' % (self.__class__.__name__, self.tp_mti, self.tp_rd, self.tp_vpf, self.tp_rp, self.tp_udhi, self.tp_srr, self.tp_da, self.tp_pid, self.tp_dcs, self.tp_vp, self.tp_udl, self.tp_ud) @classmethod - def fromBytes(cls, inb:BytesOrHex) -> 'SMS_SUBMIT': + def from_bytes(cls, inb:BytesOrHex) -> 'SMS_SUBMIT': """Construct a SMS_SUBMIT instance from the binary encoded format as used in T-PDU.""" offset = 0 if isinstance(inb, str): @@ -285,7 +284,7 @@ class SMS_SUBMIT(SMS_TPDU): offset += 1 d['tp_mr']= inb[offset] offset += 1 - da, remainder = AddressField.fromBytes(inb[2:]) + da, remainder = AddressField.from_bytes(inb[2:]) d['tp_da'] = da offset = 0 @@ -303,12 +302,10 @@ class SMS_SUBMIT(SMS_TPDU): # TODO: further decode d['tp_vp'] = remainder[offset:offset+7] offset += 7 - pass elif d['tp_vpf'] == 'absolute': # TODO: further decode d['tp_vp'] = remainder[offset:offset+7] offset += 7 - pass else: raise ValueError('Invalid VPF: %s' % d['tp_vpf']) d['tp_udl'] = remainder[offset] @@ -316,7 +313,7 @@ class SMS_SUBMIT(SMS_TPDU): d['tp_ud'] = remainder[offset:] return cls(**d) - def toBytes(self) -> bytes: + def to_bytes(self) -> bytes: """Encode a SMS_SUBMIT instance to the binary encoded format as used in T-PDU.""" outb = bytearray() d = { @@ -326,7 +323,7 @@ class SMS_SUBMIT(SMS_TPDU): flags = SMS_SUBMIT.flags_construct.build(d) outb.extend(flags) outb.append(self.tp_mr) - outb.extend(self.tp_da.toBytes()) + outb.extend(self.tp_da.to_bytes()) outb.append(self.tp_pid) outb.append(self.tp_dcs) if self.tp_vpf != 'none': @@ -336,20 +333,20 @@ class SMS_SUBMIT(SMS_TPDU): return outb @classmethod - def fromSmpp(cls, smpp_pdu) -> 'SMS_SUBMIT': + def from_smpp(cls, smpp_pdu) -> 'SMS_SUBMIT': """Construct a SMS_SUBMIT instance from the format used by smpp.pdu.""" if smpp_pdu.id == pdu_types.CommandId.submit_sm: - return cls.fromSmppSubmit(smpp_pdu) + return cls.from_smpp_submit(smpp_pdu) else: raise ValueError('Unsupported SMPP commandId %s' % smpp_pdu.id) @classmethod - def fromSmppSubmit(cls, smpp_pdu) -> 'SMS_SUBMIT': + def from_smpp_submit(cls, smpp_pdu) -> 'SMS_SUBMIT': """Construct a SMS_SUBMIT instance from the submit format used by smpp.pdu.""" ensure_smpp_is_8bit(smpp_pdu.params['data_coding']) - tp_da = AddressField.fromSmpp(smpp_pdu.params['destination_addr'], - smpp_pdu.params['dest_addr_ton'], - smpp_pdu.params['dest_addr_npi']) + tp_da = AddressField.from_smpp(smpp_pdu.params['destination_addr'], + smpp_pdu.params['dest_addr_ton'], + smpp_pdu.params['dest_addr_npi']) tp_ud = smpp_pdu.params['short_message'] #vp_smpp = smpp_pdu.params['validity_period'] #if not vp_smpp: @@ -370,7 +367,7 @@ class SMS_SUBMIT(SMS_TPDU): } return cls(**d) - def toSmpp(self) -> pdu_types.PDU: + def to_smpp(self) -> pdu_types.PDU: """Translate a SMS_SUBMIT instance to a smpp.pdu.operations.SubmitSM instance.""" esm_class = pdu_types.EsmClass(pdu_types.EsmClassMode.DEFAULT, pdu_types.EsmClassType.DEFAULT) reg_del = pdu_types.RegisteredDelivery(pdu_types.RegisteredDeliveryReceipt.NO_SMSC_DELIVERY_RECEIPT_REQUESTED) @@ -382,7 +379,7 @@ class SMS_SUBMIT(SMS_TPDU): if self.tp_dcs != 0xF6: raise ValueError('Unsupported DCS: We only support DCS=0xF6 for now') dc = pdu_types.DataCoding(pdu_types.DataCodingScheme.DEFAULT, pdu_types.DataCodingDefault.OCTET_UNSPECIFIED) - (daddr, ton, npi) = self.tp_da.toSmpp() + (daddr, ton, npi) = self.tp_da.to_smpp() return operations.SubmitSM(service_type='', source_addr_ton=pdu_types.AddrTon.ALPHANUMERIC, source_addr_npi=pdu_types.AddrNpi.UNKNOWN, diff --git a/tests/test_ota.py b/tests/test_ota.py index 7de5cff1..ce10a7d4 100644 --- a/tests/test_ota.py +++ b/tests/test_ota.py @@ -270,8 +270,8 @@ class SmsOtaTestCase(OtaTestCase): tpdu = SMS_DELIVER(tp_udhi=True, tp_oa=self.da, tp_pid=0x7F, tp_dcs=0xF6, tp_scts=h2b('22705200000000'), tp_udl=3, tp_ud=with_udh) #print("TPDU: %s" % tpdu) - #print("tpdu: %s" % b2h(tpdu.toBytes())) - self.assertEqual(b2h(tpdu.toBytes()), t['request']['encoded_tpdu']) + #print("tpdu: %s" % b2h(tpdu.to_bytes())) + self.assertEqual(b2h(tpdu.to_bytes()), t['request']['encoded_tpdu']) def test_decode_resp(self): for t in SmsOtaTestCase.testdatasets: diff --git a/tests/test_sms.py b/tests/test_sms.py index 83552245..71303996 100644 --- a/tests/test_sms.py +++ b/tests/test_sms.py @@ -6,7 +6,7 @@ from pySim.sms import * class Test_SMS_UDH(unittest.TestCase): def test_single_ie(self): - udh, tail = UserDataHeader.fromBytes('027100') + udh, tail = UserDataHeader.from_bytes('027100') self.assertEqual(len(udh.ies), 1) ie = udh.ies[0] self.assertEqual(ie.iei, 0x71) @@ -15,7 +15,7 @@ class Test_SMS_UDH(unittest.TestCase): self.assertEqual(tail, b'') def test_single_ie_tail(self): - udh, tail = UserDataHeader.fromBytes('027100abcdef') + udh, tail = UserDataHeader.from_bytes('027100abcdef') self.assertEqual(len(udh.ies), 1) ie = udh.ies[0] self.assertEqual(ie.iei, 0x71) @@ -24,7 +24,7 @@ class Test_SMS_UDH(unittest.TestCase): self.assertEqual(tail, b'\xab\xcd\xef') def test_single_ie_value(self): - udh, tail = UserDataHeader.fromBytes('03710110') + udh, tail = UserDataHeader.from_bytes('03710110') self.assertEqual(len(udh.ies), 1) ie = udh.ies[0] self.assertEqual(ie.iei, 0x71) @@ -33,7 +33,7 @@ class Test_SMS_UDH(unittest.TestCase): self.assertEqual(tail, b'') def test_two_ie_data_tail(self): - udh, tail = UserDataHeader.fromBytes('0571007001ffabcd') + udh, tail = UserDataHeader.from_bytes('0571007001ffabcd') self.assertEqual(len(udh.ies), 2) ie = udh.ies[0] self.assertEqual(ie.iei, 0x71) @@ -45,42 +45,42 @@ class Test_SMS_UDH(unittest.TestCase): self.assertEqual(ie.value, b'\xff') self.assertEqual(tail, b'\xab\xcd') - def test_toBytes(self): + def test_to_bytes(self): indata = h2b('0571007001ff') - udh, tail = UserDataHeader.fromBytes(indata) - encoded = udh.toBytes() + udh, tail = UserDataHeader.from_bytes(indata) + encoded = udh.to_bytes() self.assertEqual(encoded, indata) class Test_AddressField(unittest.TestCase): - def test_fromBytes(self): + def test_from_bytes(self): encoded = h2b('0480214399') - af, trailer = AddressField.fromBytes(encoded) + af, trailer = AddressField.from_bytes(encoded) self.assertEqual(trailer, b'\x99') self.assertEqual(af.ton, 'unknown') self.assertEqual(af.npi, 'unknown') self.assertEqual(af.digits, '1234') - def test_fromBytes_odd(self): - af, trailer = AddressField.fromBytes('038021f399') + def test_from_bytes_odd(self): + af, trailer = AddressField.from_bytes('038021f399') self.assertEqual(trailer, b'\x99') self.assertEqual(af.ton, 'unknown') self.assertEqual(af.npi, 'unknown') self.assertEqual(af.digits, '123') - def test_toBytes(self): + def test_to_bytes(self): encoded = h2b('04802143') - af, trailer = AddressField.fromBytes(encoded) - self.assertEqual(af.toBytes(), encoded) + af, trailer = AddressField.from_bytes(encoded) + self.assertEqual(af.to_bytes(), encoded) - def test_toBytes_odd(self): + def test_to_bytes_odd(self): af = AddressField('12345', 'international', 'isdn_e164') - encoded = af.toBytes() + encoded = af.to_bytes() self.assertEqual(encoded, h2b('05912143f5')) class Test_SUBMIT(unittest.TestCase): - def test_fromBytes(self): - s = SMS_SUBMIT.fromBytes('550d0b911614261771f000f5a78c0b050423f423f40003010201424547494e3a56434152440d0a56455253494f4e3a322e310d0a4e3a4d650d0a54454c3b505245463b43454c4c3b564f4943453a2b36313431363237313137300d0a54454c3b484f4d453b564f4943453a2b36313339353337303437310d0a54454c3b574f524b3b564f4943453a2b36313339363734373031350d0a454e443a') + def test_from_bytes(self): + s = SMS_SUBMIT.from_bytes('550d0b911614261771f000f5a78c0b050423f423f40003010201424547494e3a56434152440d0a56455253494f4e3a322e310d0a4e3a4d650d0a54454c3b505245463b43454c4c3b564f4943453a2b36313431363237313137300d0a54454c3b484f4d453b564f4943453a2b36313339353337303437310d0a54454c3b574f524b3b564f4943453a2b36313339363734373031350d0a454e443a') self.assertEqual(s.tp_mti, 1) self.assertEqual(s.tp_rd, True) self.assertEqual(s.tp_vpf, 'relative') @@ -92,8 +92,8 @@ class Test_SUBMIT(unittest.TestCase): self.assertEqual(s.tp_udl, 140) class Test_DELIVER(unittest.TestCase): - def test_fromBytes(self): - d = SMS_DELIVER.fromBytes('0408D0E5759A0E7FF6907090307513000824010101BB400101') + def test_from_bytes(self): + d = SMS_DELIVER.from_bytes('0408D0E5759A0E7FF6907090307513000824010101BB400101') self.assertEqual(d.tp_mti, 0) self.assertEqual(d.tp_mms, True) self.assertEqual(d.tp_lp, False)