bip: refuse OPEN CHANNEL with a TERMINAL RESPONSE

handle_OpenChannel() raises where TS 102 223 demands a TERMINAL RESPONSE.
Raising takes the whole proactive session down with it.

All four are a BIP errors and differ only in the cause byte of 8.12.11.

Change-Id: If8f01e6acb9cb3f7af952a0760fc093db81be53a
This commit is contained in:
Eric Wild
2026-09-23 17:51:04 +02:00
parent eb8e40948b
commit d671cee649
2 changed files with 108 additions and 11 deletions
+36 -9
View File
@@ -374,26 +374,53 @@ class Proact(ProactiveHandler):
other_addr_ie = Proact._find_first_element_of_type(pcmd.children, OtherAddress)
bearer_desc_ie = Proact._find_first_element_of_type(pcmd.children, BearerDescription)
buffer_size_ie = Proact._find_first_element_of_type(pcmd.children, BufferSize)
if transp_lvl_ie.decoded['protocol_type'] != 'tcp_uicc_client_remote':
raise ValueError('Unsupported protocol_type')
if other_addr_ie.decoded.get('type_of_address', None) != 'ipv4':
raise ValueError('Unsupported type_of_address')
def refuse(additional_information: str, chan_nr: int = 0):
"""TERMINAL RESPONSE refusing the OPEN CHANNEL
- always a BIP error, only the cause byte of TS 102 223 8.12.11 differs
- chan_nr 0 -> "no channel available" in the Channel status, 8.56
- 6.8.18, 6.8.20, 6.8.21 want chan status, Bearer desc and buf size
in a successful or unsuccessful response
"""
ies = [ChannelStatus(decoded=self._channel_status(chan_nr, established=False))]
ies += [ie for ie in (bearer_desc_ie, buffer_size_ie) if ie is not None]
return self._bip_response_head(pcmd, 'bearer_independent_protocol_error',
additional_information) + ies
# UICC/terminal interface transport level is Optional, TS 102 223 6.6.27.x. Absent means
# the CAT application runs its own network and transport layer, which we do not do.
if transp_lvl_ie is None or transp_lvl_ie.decoded['protocol_type'] != 'tcp_uicc_client_remote':
logger.warning("OpenChannel: unsupported UICC/terminal interface transport level (%s) "
"-> refusing", transp_lvl_ie.decoded if transp_lvl_ie else '(absent)')
return refuse('requested_uicc_if_transp_level_not_available')
if other_addr_ie is None or other_addr_ie.decoded.get('type_of_address', None) != 'ipv4':
# No cause byte fits a wrong address family. '06' is about the transport level data
# object, and 8.12.11 leaves '14' ("IPv4 only allowed") reserved by 3GPP, so '00'.
logger.warning("OpenChannel: unsupported data destination address (%s) -> refusing",
other_addr_ie.decoded if other_addr_ie else '(absent)')
return refuse('no_specific_cause')
addr_bytes = h2b(other_addr_ie.decoded['address']) if isinstance(
other_addr_ie.decoded['address'], str) else other_addr_ie.decoded['address']
ipv4_str = '%u.%u.%u.%u' % (addr_bytes[0], addr_bytes[1], addr_bytes[2], addr_bytes[3])
port_nr = transp_lvl_ie.decoded['port_number']
logger.info("OpenChannel: connecting to %s:%u", ipv4_str, port_nr)
channel = self.channels.channel_create()
try:
channel = self.channels.channel_create()
except ValueError:
# TS 102 223 6.4.27.2 and 6.4.27.3: no channel left -> BIP error
logger.warning("OpenChannel: all %u channels are in use -> refusing",
len(self.channels.channels))
return refuse('no_channel_availabile')
# yes, blocking connect()
try:
channel.connect(ipv4_str, port_nr)
except OSError as e:
logger.warning("OpenChannel: connect to %s:%u failed: %s", ipv4_str, port_nr, e)
self.channels.channel_delete(channel.chan_nr)
return self._bip_response_head(pcmd, 'bearer_independent_protocol_error',
'channel_closed') + [
ChannelStatus(decoded=self._channel_status(channel.chan_nr, established=False)),
bearer_desc_ie, buffer_size_ie]
# TS 102 223 6.4.30 is the only clause naming a cause for a link that could not be
# established: BIP error, channel closed. 6.4.27.4 lists no error cases at all.
return refuse('channel_closed', channel.chan_nr)
# Terminal Response example: [
# {'command_details': {'command_number': 1,
+72 -2
View File
@@ -94,6 +94,17 @@ def _open_channel(port, ip='127.0.0.1', cmd_nr=1):
])
def _open_channel_raw(extra_ies, cmd_nr=1):
"""OPEN CHANNEL with only the head data"""
return _pcmd([
CommandDetails(decoded={'command_number': cmd_nr, 'type_of_command': 'open_channel',
'command_qualifier': 3}).to_tlv(),
DeviceIdentities(decoded={'source_dev_id': 'uicc', 'dest_dev_id': 'terminal'}).to_tlv(),
BearerDescription(decoded={'bearer_type': 'default', 'bearer_parameters': b''}).to_tlv(),
BufferSize(decoded=1024).to_tlv(),
] + extra_ies)
def _send_data(payload, chan='channel_1', cmd_nr=1):
return _pcmd([
CommandDetails(decoded={'command_number': cmd_nr, 'type_of_command': 'send_data',
@@ -255,8 +266,67 @@ class BipRelayRoundTripTest(unittest.TestCase):
self.assertEqual(_first(til, ChannelDataLength).decoded, 0)
if __name__ == "__main__":
unittest.main()
class OpenChannelRefusalTest(unittest.TestCase):
"""Refusal is a TERMINAL RESPONSE, not an exception, raising takes the whole
proactive session down and leaves the card wondering why"""
ADDR = OtherAddress(decoded={'type_of_address': 'ipv4', 'address': bytes([127, 0, 0, 1])})
TCP = UiccTransportLevel(decoded={'protocol_type': 'tcp_uicc_client_remote', 'port_number': 1234})
def setUp(self):
self.proact = Proact()
self.addCleanup(self._close_all_channels)
def _close_all_channels(self):
for chan in list(self.proact.channels.channels.values()):
try:
chan.close()
except Exception:
pass
def _assert_refused(self, til, additional_information, chan_nr=0):
b''.join(x.to_tlv() for x in til) # must serialise, the transport posts it
res = _first(til, Result).decoded
self.assertEqual(res['general_result'], 'bearer_independent_protocol_error')
self.assertEqual(res['additional_information'], additional_information)
self.assertEqual(_first(til, ChannelStatus).decoded, '%02x00' % chan_nr) # 8.56
self.assertIsNotNone(_first(til, BearerDescription)) # 6.8.20
self.assertIsNotNone(_first(til, BufferSize)) # 6.8.21
self.assertEqual(b2h(_first(til, DeviceIdentities).to_tlv()), '82028281') # 6.8.2
def test_transport_level(self):
cases = [[self.ADDR.to_tlv()]] # absent, 6.6.27.x Optional
for proto in ('udp_uicc_client_remote', 'tcp_uicc_server', 'udp_uicc_client_local',
'tcp_uicc_client_local', 'direct_channel'): # not TCP client remote
tl = UiccTransportLevel(decoded={'protocol_type': proto, 'port_number': 1234})
cases.append([tl.to_tlv(), self.ADDR.to_tlv()])
for extra in cases:
with self.subTest(extra=b2h(extra[0])):
self._assert_refused(self.proact.handle_OpenChannel(_open_channel_raw(extra)),
'requested_uicc_if_transp_level_not_available')
def test_destination_address(self):
v6 = OtherAddress(decoded={'type_of_address': 'ipv6', 'address': bytes(16)})
for extra in ([self.TCP.to_tlv()], # absent
[self.TCP.to_tlv(), v6.to_tlv()]): # not IPv4
with self.subTest(extra=len(extra)):
self._assert_refused(self.proact.handle_OpenChannel(_open_channel_raw(extra)),
'no_specific_cause')
def test_no_channel_left(self):
for _ in range(7): # 6.4.27.2, 6.4.27.3
self.proact.channels.channel_create()
cmd = _open_channel_raw([self.TCP.to_tlv(), self.ADDR.to_tlv()])
self._assert_refused(self.proact.handle_OpenChannel(cmd), 'no_channel_availabile')
def test_connect_failure(self):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # port nothing listens on
s.bind(('127.0.0.1', 0))
dead_port = s.getsockname()[1]
s.close()
til = self.proact.handle_OpenChannel(_open_channel(dead_port))
self._assert_refused(til, 'channel_closed', chan_nr=1) # 6.4.30
self.assertEqual(self.proact.channels.channels, {}) # channel given back
class BipSinkTest(unittest.TestCase):