Hoernchen has uploaded this change for review.
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
---
M pySim/bip.py
M tests/unittests/test_bip_relay.py
2 files changed, 106 insertions(+), 11 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/63/43563/1
diff --git a/pySim/bip.py b/pySim/bip.py
index 14c75bc..8370b5b 100644
--- a/pySim/bip.py
+++ b/pySim/bip.py
@@ -356,26 +356,53 @@
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,
diff --git a/tests/unittests/test_bip_relay.py b/tests/unittests/test_bip_relay.py
index cf11b7f..8716585 100644
--- a/tests/unittests/test_bip_relay.py
+++ b/tests/unittests/test_bip_relay.py
@@ -91,6 +91,17 @@
])
+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',
@@ -252,8 +263,65 @@
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):
+ udp = UiccTransportLevel(decoded={'protocol_type': 'udp_uicc_client_remote',
+ 'port_number': 1234})
+ for extra in ([self.ADDR.to_tlv()], # absent, 6.6.27.x Optional
+ [udp.to_tlv(), self.ADDR.to_tlv()]): # not TCP client remote
+ with self.subTest(extra=len(extra)):
+ 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):
To view, visit change 43563. To unsubscribe, or for help writing mail filters, visit settings.