Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/43563?usp=email )
Change subject: bip: refuse OPEN CHANNEL with a TERMINAL RESPONSE ......................................................................
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):
-
Hoernchen