github
diff --git a/frontend/sw.js b/frontend/sw.js
index 5da1066..c61eb62 100644
--- a/frontend/sw.js
+++ b/frontend/sw.js
@@ -1,4 +1,4 @@
-const CACHE = 'otaman-v29';
+const CACHE = 'otaman-v30';
const URLS = [
'index.html',
'help.html',
diff --git a/pyproject.toml b/pyproject.toml
index 7ead484..38f9f33 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "pysim-otaman-server"
-version = "1.9.17"
+version = "1.9.18"
description = "HTTP REST server wrapping pysim for the OTAMan PWA"
requires-python = ">=3.8"
# pysim is a git-only dependency installed explicitly by setup.bat/setup.sh.
diff --git a/pysim_otaman_server/server.py b/pysim_otaman_server/server.py
index ae5d8c9..88178f1 100644
--- a/pysim_otaman_server/server.py
+++ b/pysim_otaman_server/server.py
@@ -18,7 +18,9 @@ from osmocom.construct import GsmOrUcs2Adapter
from osmocom.tlv import BER_TLV_IE
-VERSION = '1.9.17'
+VERSION = '1.9.18'
+
+MAX_ENVELOPE_SEGMENTS = 5 # max SMS segments for outgoing C-APDU in ENVELOPE
# Static file serving (the PWA lives in
/frontend, served by this server
@@ -250,7 +252,19 @@ def _send_envelope(tpdu_hex, scc, sm_sc='12345678912', submit_handler=None):
elif sw.startswith('91'):
def _capture_sms_tpdu(raw, cmd_num, cmd_type, dev_src, dev_dst):
if submit_handler:
- submit_handler.submit_tpdu_hex = _find_sms_tpdu(raw)
+ tpdu_hex = _find_sms_tpdu(raw)
+ if tpdu_hex:
+ ref, total, num, payload = _parse_sms_concat(bytes.fromhex(tpdu_hex))
+ if total is not None and num is not None:
+ submit_handler.sms_segments.append((ref, total, num, payload.hex()))
+ matching = [s for s in submit_handler.sms_segments if s[0] == ref]
+ if len(matching) >= total:
+ sorted_segs = sorted(matching, key=lambda s: s[2])
+ assembled = b''.join(bytes.fromhex(s[3]) for s in sorted_segs)
+ submit_handler.submit_tpdu_hex = assembled.hex()
+ sys.stderr.write('SMS concat: assembled %d segments (ref=%s)\n' % (total, ref))
+ else:
+ submit_handler.submit_tpdu_hex = tpdu_hex
_handle_proactive_chain(scc, sw, _capture_sms_tpdu)
data, sw = '', '9000'
if sw == '9000' and submit_handler and not submit_handler.submit_tpdu_hex:
@@ -413,10 +427,13 @@ class PoRSubmitHandler(ProactiveHandler):
"""Captures the SMS-SUBMIT TPDU from a SendShortMessage proactive command
issued by the SIM in response to PoR-in-submit (SPI2 bit 0x20).
The 91XX path in _send_envelope scans the FETCH response directly for
- the SMS_TPDU child (tag 0x8B) and populates submit_tpdu_hex."""
+ the SMS_TPDU child (tag 0x8B) and populates submit_tpdu_hex.
+ Supports concatenated SMS: when UDH contains IEI 0x00 or 0x08, segments
+ are accumulated and reassembled once all parts arrive."""
def __init__(self):
super().__init__()
self.submit_tpdu_hex = None
+ self.sms_segments = [] # [(ref, total, num, payload_hex), ...]
class _DefaultProactiveHandler(ProactiveHandler):
@@ -974,6 +991,85 @@ def _find_sms_tpdu(raw):
return None
+def _calc_ud_offset(tpdu):
+ """Calculate the offset of TP-UD (User Data) within an SMS TPDU.
+ Handles SMS-SUBMIT (MTI=01) and SMS-DELIVER (MTI=00)."""
+ first_octet = tpdu[0]
+ mti = first_octet & 0x03
+ vpf = (first_octet >> 3) & 0x03
+ if mti == 0x01:
+ # SMS-SUBMIT: 1 + 1(MR) + 1(DA_len) + 1(DA_type) + ceil(DA_len/2) + 1(PID) + 1(DCS) [+7 if VPF]
+ if len(tpdu) < 3:
+ return None
+ da_len_digits = tpdu[2]
+ da_data_bytes = (da_len_digits + 1) // 2
+ off = 1 + 1 + 1 + 1 + da_data_bytes + 1 + 1
+ if vpf in (0x01, 0x02): # relative or absolute
+ off += 7
+ if off >= len(tpdu):
+ return None
+ return off + 1 # skip UDL byte
+ elif mti == 0x00:
+ # SMS-DELIVER: 1 + 1(OA_len) + ceil(OA_len/2) + 1(PID) + 1(DCS) + 7(SCTS)
+ if len(tpdu) < 2:
+ return None
+ oa_len_digits = tpdu[1]
+ oa_data_bytes = (oa_len_digits + 1) // 2
+ off = 1 + 1 + oa_data_bytes + 1 + 1 + 7
+ if off >= len(tpdu):
+ return None
+ return off + 1 # skip UDL byte
+ return None
+
+
+def _parse_sms_concat(tpdu_bytes):
+ """Parse an SMS TPDU for UDH concatenation info and payload.
+ Returns (ref, total, num, payload_bytes) or (None, None, None, payload_bytes) if no concat."""
+ if not tpdu_bytes or len(tpdu_bytes) < 2:
+ return None, None, None, tpdu_bytes or b''
+
+ first_octet = tpdu_bytes[0]
+ udhi = bool(first_octet & 0x40)
+
+ ud_offset = _calc_ud_offset(tpdu_bytes)
+ if ud_offset is None or ud_offset >= len(tpdu_bytes):
+ return None, None, None, tpdu_bytes
+
+ if not udhi:
+ # No UDH — entire UD is the payload
+ return None, None, None, tpdu_bytes[ud_offset:]
+
+ # TP-UDHI is set: UD starts with UDHL
+ udhl = tpdu_bytes[ud_offset]
+ udh_start = ud_offset + 1
+ udh_end = udh_start + udhl
+ if udh_end > len(tpdu_bytes):
+ return None, None, None, tpdu_bytes[ud_offset:]
+
+ # Walk UDH IEs looking for concatenation
+ ref = None
+ total = None
+ num = None
+ ie_off = udh_start
+ while ie_off + 2 <= udh_end:
+ iei = tpdu_bytes[ie_off]
+ iedl = tpdu_bytes[ie_off + 1]
+ if ie_off + 2 + iedl > udh_end:
+ break
+ if iei == 0x00 and iedl == 3:
+ ref = tpdu_bytes[ie_off + 2]
+ total = tpdu_bytes[ie_off + 3]
+ num = tpdu_bytes[ie_off + 4]
+ elif iei == 0x08 and iedl == 4:
+ ref = (tpdu_bytes[ie_off + 2] << 8) | tpdu_bytes[ie_off + 3]
+ total = tpdu_bytes[ie_off + 4]
+ num = tpdu_bytes[ie_off + 5]
+ ie_off += 2 + iedl
+
+ payload = tpdu_bytes[ud_offset + 1 + udhl:] # payload after UDH
+ return ref, total, num, payload
+
+
def _parse_display_text(raw):
if raw[0] == 0xD0:
off = _skip_ber_len(raw, 1)
@@ -1856,54 +1952,58 @@ class PysimHandler(BaseHTTPRequestHandler):
body.get('spi1', ''), body.get('spi2', ''), body.get('kic', ''),
body.get('kid', ''), body.get('tar', ''), body.get('cntr', ''),
len(sp_bytes), total))
- sys.stderr.write('RAM C-APDU: %s\n' % apdu if apdu else sp)
- sys.stderr.write('RAM SECURED-PACKET: %s\n' % sp_hex)
- last_data = None
- last_sw = None
- for i, chunk in enumerate(chunks):
- tpdu = _build_sms_tpdu(chunk.hex(), total, i + 1, oa_number=self.server.sms_oa,
- include_cpi=include_cpi)
- data, sw = _send_envelope(tpdu, scc, sm_sc=self.server.sms_sc, submit_handler=submit_handler)
- last_data = data
- last_sw = sw
- if sw != '9000' and not sw.startswith('91'):
- resp = {'success': False, 'sw': sw, 'error': 'ENVELOPE failed at chunk %d' % (i + 1)}
- sys.stderr.write('OTA SEND FAILED: chunk %d SW %s\n' % (i + 1, sw))
- break
+ if total > MAX_ENVELOPE_SEGMENTS:
+ resp = {'success': False, 'error': 'Secured packet too large: %d segments (max %d)' % (total, MAX_ENVELOPE_SEGMENTS)}
+ sys.stderr.write('OTA SEND FAILED: %d segments exceeds max %d\n' % (total, MAX_ENVELOPE_SEGMENTS))
else:
- resp = {'success': True, 'sw': last_sw, 'response_data': last_data if last_data else None}
- por_src = 'envelope'
- por_hex = resp['response_data']
- if submit_handler and submit_handler.submit_tpdu_hex:
- tpdu_b = bytes.fromhex(submit_handler.submit_tpdu_hex)
- idx = tpdu_b.find(b'\x02\x71\x00')
- if idx >= 0:
- por_hex = tpdu_b[idx:].hex()
- por_src = 'sms-submit'
- por = _decode_por(body.get('spi1', ''), body.get('spi2', ''), body.get('kic', ''),
- body.get('kid', ''), body.get('cntr', ''), body.get('kicKey', ''),
- body.get('kidKey', ''), por_hex)
- # Check for SPI2=0x21 (PoR required) but got 9000 with no PoR → card refuses PoR
- is_ram = bool(apdu)
- por_required = bool(spi2_val & 0x01)
- no_por_received = not por_hex and not (submit_handler and submit_handler.submit_tpdu_hex)
- if is_ram and por_required and last_sw == '9000' and no_por_received:
- sys.stderr.write('WARNING: Card refused to return PoR - ENVELOPE returned 9000 with no response data\n')
- sys.stderr.write('RAM RESPONSE-PACKET: %s\n' % (por_hex if por_hex else 'empty'))
- if por:
- resp['por'] = por
- extra = ''
- if por.get('decoded'):
- extra = ' (compact: %s cmd, last SW %s)' % (por['decoded'].get('number_of_commands', '?'),
- por['decoded'].get('last_status_word', '?'))
- sys.stderr.write('RAM R-APDU: %s\n' % por['decoded'].get('last_response_data', ''))
- sys.stderr.write('OTA PoR[%s]: status=%s TAR=%s CNTR=%s PCNTR=%s RPL=%s RHL=%s%s\n' % (
- por_src, por.get('response_status'), por.get('tar'), por.get('cntr'),
- por.get('pcntr'), por.get('rpl'), por.get('rhl'), extra))
- elif por_hex:
- sys.stderr.write('OTA PoR[%s]: undecodable raw=%s\n' % (por_src, str(por_hex)))
+ sys.stderr.write('RAM C-APDU: %s\n' % apdu if apdu else sp)
+ sys.stderr.write('RAM SECURED-PACKET: %s\n' % sp_hex)
+ last_data = None
+ last_sw = None
+ for i, chunk in enumerate(chunks):
+ tpdu = _build_sms_tpdu(chunk.hex(), total, i + 1, oa_number=self.server.sms_oa,
+ include_cpi=include_cpi)
+ data, sw = _send_envelope(tpdu, scc, sm_sc=self.server.sms_sc, submit_handler=submit_handler)
+ last_data = data
+ last_sw = sw
+ if sw != '9000' and not sw.startswith('91'):
+ resp = {'success': False, 'sw': sw, 'error': 'ENVELOPE failed at chunk %d' % (i + 1)}
+ sys.stderr.write('OTA SEND FAILED: chunk %d SW %s\n' % (i + 1, sw))
+ break
else:
- sys.stderr.write('OTA PoR[%s]: none\n' % por_src)
+ resp = {'success': True, 'sw': last_sw, 'response_data': last_data if last_data else None}
+ por_src = 'envelope'
+ por_hex = resp['response_data']
+ if submit_handler and submit_handler.submit_tpdu_hex:
+ tpdu_b = bytes.fromhex(submit_handler.submit_tpdu_hex)
+ idx = tpdu_b.find(b'\x02\x71\x00')
+ if idx >= 0:
+ por_hex = tpdu_b[idx:].hex()
+ por_src = 'sms-submit'
+ por = _decode_por(body.get('spi1', ''), body.get('spi2', ''), body.get('kic', ''),
+ body.get('kid', ''), body.get('cntr', ''), body.get('kicKey', ''),
+ body.get('kidKey', ''), por_hex)
+ # Check for SPI2=0x21 (PoR required) but got 9000 with no PoR → card refuses PoR
+ is_ram = bool(apdu)
+ por_required = bool(spi2_val & 0x01)
+ no_por_received = not por_hex and not (submit_handler and submit_handler.submit_tpdu_hex)
+ if is_ram and por_required and last_sw == '9000' and no_por_received:
+ sys.stderr.write('WARNING: Card refused to return PoR - ENVELOPE returned 9000 with no response data\n')
+ sys.stderr.write('RAM RESPONSE-PACKET: %s\n' % (por_hex if por_hex else 'empty'))
+ if por:
+ resp['por'] = por
+ extra = ''
+ if por.get('decoded'):
+ extra = ' (compact: %s cmd, last SW %s)' % (por['decoded'].get('number_of_commands', '?'),
+ por['decoded'].get('last_status_word', '?'))
+ sys.stderr.write('RAM R-APDU: %s\n' % por['decoded'].get('last_response_data', ''))
+ sys.stderr.write('OTA PoR[%s]: status=%s TAR=%s CNTR=%s PCNTR=%s RPL=%s RHL=%s%s\n' % (
+ por_src, por.get('response_status'), por.get('tar'), por.get('cntr'),
+ por.get('pcntr'), por.get('rpl'), por.get('rhl'), extra))
+ elif por_hex:
+ sys.stderr.write('OTA PoR[%s]: undecodable raw=%s\n' % (por_src, str(por_hex)))
+ else:
+ sys.stderr.write('OTA PoR[%s]: none\n' % por_src)
finally:
if submit_handler and hasattr(scc, '_tp'):
scc._tp.proactive_handler = old_proactive
diff --git a/tests/test_ota_helpers.py b/tests/test_ota_helpers.py
index 582c6b8..fd60e3f 100644
--- a/tests/test_ota_helpers.py
+++ b/tests/test_ota_helpers.py
@@ -418,5 +418,230 @@ class TestExpandedRemoteResponse(unittest.TestCase):
self.assertEqual(entry['tr_result_name'], 'Command performed successfully')
+class TestSmsConcat(unittest.TestCase):
+ """Tests for _parse_sms_concat — SMS UDH concatenation parsing."""
+
+ def test_no_udh_sms_submit(self):
+ """SMS-SUBMIT without TP-UDHI: entire UD is payload."""
+ from pysim_otaman_server.server import _parse_sms_concat
+ # First octet 0x01: MTI=01 (SUBMIT), no UDH, no VP
+ # MR=00, DA_len=05, DA_type=90, DA=2143F5, PID=00, DCS=04, UDL=03, UD=AABBCC
+ tpdu = bytes.fromhex('0100' # first octet + MR
+ '05' # DA length
+ '90' # DA type
+ '2143F5' # DA data (3 bytes for 5 digits)
+ '0004' # PID + DCS
+ '03' # UDL
+ 'AABBCC') # UD (payload)
+ ref, total, num, payload = _parse_sms_concat(tpdu)
+ self.assertIsNone(ref)
+ self.assertIsNone(total)
+ self.assertIsNone(num)
+ self.assertEqual(payload.hex(), 'aabbcc')
+
+ def test_8bit_concat_iei_0x00(self):
+ """SMS-SUBMIT with IEI 0x00 (8-bit reference concatenation)."""
+ from pysim_otaman_server.server import _parse_sms_concat
+ # First octet 0x41: MTI=01 (SUBMIT), TP-UDHI=1, no VP
+ # MR=00, DA_len=05, DA_type=90, DA=2143F5, PID=00, DCS=04
+ # UDL=09, UDHL=05, UDH: 00 03 04 04 01 (concat IE), payload=AABBCC
+ tpdu = bytes.fromhex('4100' # first octet + MR
+ '05' # DA length
+ '90' # DA type
+ '2143F5' # DA data
+ '0004' # PID + DCS
+ '09' # UDL (1 UDHL + 5 UDH + 3 payload = 9)
+ '05' # UDHL = 5 bytes of UDH
+ '0003' # IEI=0x00, IEDL=3
+ '04' # ref
+ '04' # total (4 segments)
+ '01' # num (segment 1)
+ 'AABBCC') # payload
+ ref, total, num, payload = _parse_sms_concat(tpdu)
+ self.assertEqual(ref, 0x04)
+ self.assertEqual(total, 4)
+ self.assertEqual(num, 1)
+ self.assertEqual(payload.hex(), 'aabbcc')
+
+ def test_16bit_concat_iei_0x08(self):
+ """SMS-SUBMIT with IEI 0x08 (16-bit reference concatenation)."""
+ from pysim_otaman_server.server import _parse_sms_concat
+ # First octet 0x41: MTI=01, TP-UDHI=1
+ # UDH: 06 (UDHL) 08 04 01 02 03 04 (16-bit concat: ref=0x0102, total=3, num=4)
+ # payload=FF
+ tpdu = bytes.fromhex('4100'
+ '05'
+ '90'
+ '2143F5'
+ '0004'
+ '08' # UDL (1 UDHL + 6 UDH + 1 payload = 8)
+ '06' # UDHL
+ '0804' # IEI=0x08, IEDL=4
+ '0102' # ref (16-bit, big-endian)
+ '03' # total
+ '04' # num
+ 'FF') # payload
+ ref, total, num, payload = _parse_sms_concat(tpdu)
+ self.assertEqual(ref, 0x0102)
+ self.assertEqual(total, 3)
+ self.assertEqual(num, 4)
+ self.assertEqual(payload.hex(), 'ff')
+
+ def test_udh_with_cpi(self):
+ """UDH with concatenation IE + CPI IE (0x70)."""
+ from pysim_otaman_server.server import _parse_sms_concat
+ # First octet 0x41: MTI=01, TP-UDHI=1
+ # UDHL=07, UDH: 00 03 04 04 01 (concat) + 70 00 (CPI)
+ tpdu = bytes.fromhex('4100'
+ '05'
+ '90'
+ '2143F5'
+ '0004'
+ '0A' # UDL (1 UDHL + 7 UDH + 1 payload = 9? no: 1+5+2+1=9, but UDH=7 bytes)
+ '07' # UDHL = 7
+ '0003' # IEI=0x00, IEDL=3
+ '04' # ref
+ '04' # total
+ '01' # num
+ '7000' # CPI IE (IEI=0x70, IEDL=0)
+ 'DD') # payload
+ ref, total, num, payload = _parse_sms_concat(tpdu)
+ self.assertEqual(ref, 0x04)
+ self.assertEqual(total, 4)
+ self.assertEqual(num, 1)
+ self.assertEqual(payload.hex(), 'dd')
+
+ def test_empty_payload(self):
+ """Segment with empty payload after UDH."""
+ from pysim_otaman_server.server import _parse_sms_concat
+ # First octet 0x41: MTI=01, TP-UDHI=1
+ tpdu = bytes.fromhex('4100'
+ '05'
+ '90'
+ '2143F5'
+ '0004'
+ '06' # UDL (1 UDHL + 5 UDH + 0 payload = 6)
+ '05' # UDHL
+ '0003'
+ '01'
+ '02'
+ '01') # no payload after UDH
+ ref, total, num, payload = _parse_sms_concat(tpdu)
+ self.assertEqual(ref, 0x01)
+ self.assertEqual(total, 2)
+ self.assertEqual(num, 1)
+ self.assertEqual(len(payload), 0)
+
+ def test_short_tpdu(self):
+ """Truncated TPDU returns gracefully."""
+ from pysim_otaman_server.server import _parse_sms_concat
+ ref, total, num, payload = _parse_sms_concat(b'\x01')
+ self.assertIsNone(ref)
+ self.assertIsNone(total)
+ self.assertIsNone(num)
+
+ def test_none_input(self):
+ """None input returns empty payload."""
+ from pysim_otaman_server.server import _parse_sms_concat
+ ref, total, num, payload = _parse_sms_concat(None)
+ self.assertIsNone(ref)
+ self.assertIsNone(total)
+ self.assertIsNone(num)
+ self.assertEqual(len(payload), 0)
+
+ def test_short_tpdu(self):
+ """Truncated TPDU returns gracefully."""
+ from pysim_otaman_server.server import _parse_sms_concat
+ ref, total, num, payload = _parse_sms_concat(b'\x44')
+ self.assertIsNone(ref)
+ self.assertIsNone(total)
+ self.assertIsNone(num)
+
+ def test_none_input(self):
+ """None input returns empty payload."""
+ from pysim_otaman_server.server import _parse_sms_concat
+ ref, total, num, payload = _parse_sms_concat(None)
+ self.assertIsNone(ref)
+ self.assertIsNone(total)
+ self.assertIsNone(num)
+ self.assertEqual(len(payload), 0)
+
+
+class TestSmsReassembly(unittest.TestCase):
+ """Tests for SMS segment reassembly logic."""
+
+ def test_single_segment_no_concat(self):
+ """Single segment without UDH → submit_tpdu_hex is set directly."""
+ from pysim_otaman_server.server import PoRSubmitHandler, _find_sms_tpdu, _parse_sms_concat
+ handler = PoRSubmitHandler()
+ # Build a simple D0 with tag 8B containing an SMS-SUBMIT without UDH
+ sms_tpdu = bytes.fromhex('040005902143F50004' # SMS-SUBMIT header
+ '03' # UDL
+ 'AABBCC') # payload
+ # Wrap in D0 proactive command
+ d0 = bytes([0xD0, len(sms_tpdu) + 4, # approximate BER length
+ 0x81, 0x03, 0x01, 0x13, 0x00, # Command Details
+ 0x82, 0x02, 0x81, 0x83, # Device Identities
+ 0x8B, len(sms_tpdu)]) # tag 8B
+ # Simulate _find_sms_tpdu extracting tag 8B
+ found = sms_tpdu.hex()
+ # Parse and check
+ ref, total, num, payload = _parse_sms_concat(sms_tpdu)
+ self.assertIsNone(ref)
+ handler.submit_tpdu_hex = found # single segment path
+ self.assertEqual(handler.submit_tpdu_hex, found)
+
+ def test_multi_segment_reassembly(self):
+ """3 segments with IEI 0x00 in random order → assembled in correct order."""
+ from pysim_otaman_server.server import PoRSubmitHandler
+ handler = PoRSubmitHandler()
+
+ # Segment payloads (after UDH)
+ payloads = [b'\x01\x02', b'\x03\x04', b'\x05\x06']
+ ref = 0x42
+ total = 3
+
+ # Simulate receiving segments in random order: 2, 0, 1
+ for idx in [1, 0, 2]:
+ num = idx + 1
+ handler.sms_segments.append((ref, total, num, payloads[idx].hex()))
+ # Check if all segments collected
+ matching = [s for s in handler.sms_segments if s[0] == ref]
+ if len(matching) >= total:
+ sorted_segs = sorted(matching, key=lambda s: s[2])
+ assembled = b''.join(bytes.fromhex(s[3]) for s in sorted_segs)
+ handler.submit_tpdu_hex = assembled.hex()
+
+ self.assertEqual(handler.submit_tpdu_hex, '010203040506')
+
+ def test_independent_references(self):
+ """Two different reference numbers are independent."""
+ from pysim_otaman_server.server import PoRSubmitHandler
+ handler = PoRSubmitHandler()
+
+ # Ref 0x01: 2 segments
+ handler.sms_segments.append((0x01, 2, 1, 'AA'))
+ handler.sms_segments.append((0x01, 2, 2, 'BB'))
+ matching = [s for s in handler.sms_segments if s[0] == 0x01]
+ if len(matching) >= 2:
+ sorted_segs = sorted(matching, key=lambda s: s[2])
+ assembled = b''.join(bytes.fromhex(s[3]) for s in sorted_segs)
+ handler.submit_tpdu_hex = assembled.hex()
+
+ self.assertEqual(handler.submit_tpdu_hex, 'aabb')
+
+ # Ref 0x02: 1 segment (independent)
+ handler.sms_segments.append((0x02, 1, 1, 'CC'))
+ matching2 = [s for s in handler.sms_segments if s[0] == 0x02]
+ if len(matching2) >= 1:
+ sorted_segs2 = sorted(matching2, key=lambda s: s[2])
+ assembled2 = b''.join(bytes.fromhex(s[3]) for s in sorted_segs2)
+ # Only update if ref 0x02 is complete
+ handler.submit_tpdu_hex = assembled2.hex()
+
+ # Last assembly was ref 0x02
+ self.assertEqual(handler.submit_tpdu_hex, 'cc')
+
+
if __name__ == '__main__':
unittest.main()