diff --git a/hb_confbridge.py b/hb_confbridge.py index 42e5e85..0b9d2f0 100755 --- a/hb_confbridge.py +++ b/hb_confbridge.py @@ -179,6 +179,13 @@ def rule_timer_loop(): if CONFIG['REPORTS']['REPORT']: report_server.send_clients('bridge updated') + + for system in CONFIG['SYSTEMS']: + if CONFIG['SYSTEMS'][system]['MODE'] == 'OPENBRIDGE': + for stream_id in system['STATUS']: + if system['STATUS'][stream_id]['STREAM_START'] < _now + 1: + removed = system['STATUS'].pop(stream_id) + logger.warning('STALE OPENBRIDGE STREAM ID REMOVED FROM SYSTEM: %s, STREAM ID %s', system, int_id(stream_id)) class routerOBP(OPENBRIDGE): @@ -198,10 +205,14 @@ class routerOBP(OPENBRIDGE): if allow_sub(_rf_src) == False: self._logger.warning('(%s) Group Voice Packet ***REJECTED BY ACL*** From: %s, HBP Peer %s, Destination TGID %s', self._system, int_id(_rf_src), int_id(_peer_id), int_id(_dst_id)) return + + # Is this a new call stream? + if (_stream_id not in self.STATUS): - if _stream_id not in self.STATUS: # This is a new call stream - self.STATUS[_stream_id] = pkt_time + self.STATUS[_stream_id]['STREAM_START'] = pkt_time + self.STATUS[_stream_id]['PKT_COUNT'] = 0 + self.STATUS[_stream_id]['CONTENTION'] = False self._logger.info('(%s) *CALL START* STREAM ID: %s SUB: %s (%s) PEER: %s (%s) TGID %s (%s), TS %s', \ self._system, int_id(_stream_id), get_alias(_rf_src, subscriber_ids), int_id(_rf_src), get_alias(_peer_id, peer_ids), int_id(_peer_id), get_alias(_dst_id, talkgroup_ids), int_id(_dst_id), _slot) if CONFIG['REPORTS']['REPORT']: @@ -210,12 +221,12 @@ class routerOBP(OPENBRIDGE): # If we can, use the LC from the voice header as to keep all options intact if _frame_type == hb_const.HBPF_DATA_SYNC and _dtype_vseq == hb_const.HBPF_SLT_VHEAD: decoded = decode.voice_head_term(dmrpkt) - self.STATUS[_slot]['RX_LC'] = decoded['LC'] + self.STATUS[_stream_id]['LC'] = decoded['LC'] # If we don't have a voice header then don't wait to decode it from the Embedded LC # just make a new one from the HBP header. This is good enough, and it saves lots of time else: - self.STATUS[_slot]['RX_LC'] = const.LC_OPT + _dst_id + _rf_src + self.STATUS[_stream_id] = const.LC_OPT + _dst_id + _rf_src for _bridge in BRIDGES: @@ -224,73 +235,98 @@ class routerOBP(OPENBRIDGE): if (_system['SYSTEM'] == self._system and _system['TGID'] == _dst_id and _system['TS'] == _slot and _system['ACTIVE'] == True): for _target in BRIDGES[_bridge]: - if _target['SYSTEM'] != self._system: - if _target['ACTIVE']: - _target_status = systems[_target['SYSTEM']].STATUS - _target_system = self._CONFIG['SYSTEMS'][_target['SYSTEM']] - - if (_stream_id != self.STATUS[_slot]['RX_STREAM_ID']) or (_target_status[_target['TS']]['TX_RFS'] != _rf_src) or (_target_status[_target['TS']]['TX_TGID'] != _target['TGID']): - # Record the DST TGID and Stream ID - _target_status[_target['TS']]['TX_TGID'] = _target['TGID'] - _target_status[_target['TS']]['TX_STREAM_ID'] = _stream_id - _target_status[_target['TS']]['TX_RFS'] = _rf_src - # Generate LCs (full and EMB) for the TX stream - dst_lc = self.STATUS[_slot]['RX_LC'][0:3] + _target['TGID'] + _rf_src - _target_status[_target['TS']]['TX_H_LC'] = bptc.encode_header_lc(dst_lc) - _target_status[_target['TS']]['TX_T_LC'] = bptc.encode_terminator_lc(dst_lc) - _target_status[_target['TS']]['TX_EMB_LC'] = bptc.encode_emblc(dst_lc) - self._logger.debug('(%s) Generating TX FULL and EMB LCs for destination: System: %s, TS: %s, TGID: %s', self._system, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) - self._logger.info('(%s) Conference Bridge: %s, Call Bridged to: System: %s TS: %s, TGID: %s', self._system, _bridge, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) - - # Handle any necessary re-writes for the destination - if _system['TS'] != _target['TS']: - _tmp_bits = _bits ^ 1 << 7 - else: - _tmp_bits = _bits - - # Assemble transmit HBP packet header - _tmp_data = _data[:8] + _target['TGID'] + _data[11:15] + chr(_tmp_bits) + _data[16:20] - - # MUST TEST FOR NEW STREAM AND IF SO, RE-WRITE THE LC FOR THE TARGET - # MUST RE-WRITE DESTINATION TGID IF DIFFERENT - # if _dst_id != rule['DST_GROUP']: - dmrbits = bitarray(endian='big') - dmrbits.frombytes(dmrpkt) - # Create a voice header packet (FULL LC) - if _frame_type == hb_const.HBPF_DATA_SYNC and _dtype_vseq == hb_const.HBPF_SLT_VHEAD: - dmrbits = _target_status[_target['TS']]['TX_H_LC'][0:98] + dmrbits[98:166] + _target_status[_target['TS']]['TX_H_LC'][98:197] - # Create a voice terminator packet (FULL LC) - elif _frame_type == hb_const.HBPF_DATA_SYNC and _dtype_vseq == hb_const.HBPF_SLT_VTERM: - dmrbits = _target_status[_target['TS']]['TX_T_LC'][0:98] + dmrbits[98:166] + _target_status[_target['TS']]['TX_T_LC'][98:197] - # Create a Burst B-E packet (Embedded LC) - elif _dtype_vseq in [1,2,3,4]: - dmrbits = dmrbits[0:116] + _target_status[_target['TS']]['TX_EMB_LC'][_dtype_vseq] + dmrbits[148:264] - dmrpkt = dmrbits.tobytes() - _tmp_data = _tmp_data + dmrpkt + _data[53:55] - - # Transmit the packet to the destination system - systems[_target['SYSTEM']].send_system(_tmp_data) - #self._logger.debug('(%s) Packet routed by bridge: %s to system: %s TS: %s, TGID: %s', self._system, _bridge, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) + if (_target['SYSTEM'] != self._system) and (_target['ACTIVE']) and (_target['MODE'] != 'OPENBRIDGE'): + _target_status = systems[_target['SYSTEM']].STATUS + _target_system = self._CONFIG['SYSTEMS'][_target['SYSTEM']] + + # BEGIN CONTENTION HANDLING + # + # The rules for each of the 4 "ifs" below are listed here for readability. The Frame To Send is: + # From a different group than last RX from this HBSystem, but it has been less than Group Hangtime + # From a different group than last TX to this HBSystem, but it has been less than Group Hangtime + # From the same group as the last RX from this HBSystem, but from a different subscriber, and it has been less than stream timeout + # From the same group as the last TX to this HBSystem, but from a different subscriber, and it has been less than stream timeout + # The "continue" at the end of each means the next iteration of the for loop that tests for matching rules + # + if ((_target['TGID'] != _target_status[_target['TS']]['RX_TGID']) and ((pkt_time - _target_status[_target['TS']]['RX_TIME']) < _target_system['GROUP_HANGTIME'])): + if self.STATUS[_stream_id]['CONTENTION'] == False: + self.STATUS[_stream_id]['CONTENTION'] = True + self._logger.info('(%s) Call not routed to TGID %s, target active or in group hangtime: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_target['TGID']), _target['SYSTEM'], _target['TS'], int_id(_target_status[_target['TS']]['RX_TGID'])) + continue + if ((_target['TGID'] != _target_status[_target['TS']]['TX_TGID']) and ((pkt_time - _target_status[_target['TS']]['TX_TIME']) < _target_system['GROUP_HANGTIME'])): + if self.STATUS[_stream_id]['CONTENTION'] == False: + self.STATUS[_stream_id]['CONTENTION'] = True + self._logger.info('(%s) Call not routed to TGID%s, target in group hangtime: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_target['TGID']), _target['SYSTEM'], _target['TS'], int_id(_target_status[_target['TS']]['TX_TGID'])) + continue + if (_target['TGID'] == _target_status[_target['TS']]['RX_TGID']) and ((pkt_time - _target_status[_target['TS']]['RX_TIME']) < hb_const.STREAM_TO): + if self.STATUS[_stream_id]['CONTENTION'] == False: + self.STATUS[_stream_id]['CONTENTION'] = True + self._logger.info('(%s) Call not routed to TGID%s, matching call already active on target: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_target['TGID']), _target['SYSTEM'], _target['TS'], int_id(_target_status[_target['TS']]['RX_TGID'])) + continue + if (_target['TGID'] == _target_status[_target['TS']]['TX_TGID']) and (_rf_src != _target_status[_target['TS']]['TX_RFS']) and ((pkt_time - _target_status[_target['TS']]['TX_TIME']) < hb_const.STREAM_TO): + if self.STATUS[_stream_id]['CONTENTION'] == False: + self.STATUS[_stream_id]['CONTENTION'] = True + self._logger.info('(%s) Call not routed for subscriber %s, call route in progress on target: HBSystem: %s, TS: %s, TGID: %s, SUB: %s', self._system, int_id(_rf_src), _target['SYSTEM'], _target['TS'], int_id(_target_status[_target['TS']]['TX_TGID']), int_id(_target_status[_target['TS']]['TX_RFS'])) + continue + + # Set values for the contention handler to test next time there is a frame to forward + _target_status[_target['TS']]['TX_TIME'] = pkt_time + + if (_target_status[_target['TS']]['TX_RFS'] != _rf_src) or (_target_status[_target['TS']]['TX_TGID'] != _target['TGID']): + # Record the DST TGID and Stream ID + _target_status[_target['TS']]['TX_TGID'] = _target['TGID'] + _target_status[_target['TS']]['TX_STREAM_ID'] = _stream_id + _target_status[_target['TS']]['TX_RFS'] = _rf_src + # Generate LCs (full and EMB) for the TX stream + dst_lc = self.STATUS[_stream_id]['LC'][0:3] + _target['TGID'] + _rf_src + _target_status[_target['TS']]['TX_H_LC'] = bptc.encode_header_lc(dst_lc) + _target_status[_target['TS']]['TX_T_LC'] = bptc.encode_terminator_lc(dst_lc) + _target_status[_target['TS']]['TX_EMB_LC'] = bptc.encode_emblc(dst_lc) + self._logger.debug('(%s) Generating TX FULL and EMB LCs for destination: System: %s, TS: %s, TGID: %s', self._system, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) + self._logger.info('(%s) Conference Bridge: %s, Call Bridged to: System: %s TS: %s, TGID: %s', self._system, _bridge, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) + + # Handle any necessary re-writes for the destination + if _system['TS'] != _target['TS']: + _tmp_bits = _bits ^ 1 << 7 + else: + _tmp_bits = _bits + + # Assemble transmit HBP packet header + _tmp_data = _data[:8] + _target['TGID'] + _data[11:15] + chr(_tmp_bits) + _data[16:20] + + # MUST TEST FOR NEW STREAM AND IF SO, RE-WRITE THE LC FOR THE TARGET + # MUST RE-WRITE DESTINATION TGID IF DIFFERENT + # if _dst_id != rule['DST_GROUP']: + dmrbits = bitarray(endian='big') + dmrbits.frombytes(dmrpkt) + # Create a voice header packet (FULL LC) + if _frame_type == hb_const.HBPF_DATA_SYNC and _dtype_vseq == hb_const.HBPF_SLT_VHEAD: + dmrbits = _target_status[_target['TS']]['TX_H_LC'][0:98] + dmrbits[98:166] + _target_status[_target['TS']]['TX_H_LC'][98:197] + # Create a voice terminator packet (FULL LC) + elif _frame_type == hb_const.HBPF_DATA_SYNC and _dtype_vseq == hb_const.HBPF_SLT_VTERM: + dmrbits = _target_status[_target['TS']]['TX_T_LC'][0:98] + dmrbits[98:166] + _target_status[_target['TS']]['TX_T_LC'][98:197] + # Create a Burst B-E packet (Embedded LC) + elif _dtype_vseq in [1,2,3,4]: + dmrbits = dmrbits[0:116] + _target_status[_target['TS']]['TX_EMB_LC'][_dtype_vseq] + dmrbits[148:264] + dmrpkt = dmrbits.tobytes() + _tmp_data = _tmp_data + dmrpkt + _data[53:55] + + # Transmit the packet to the destination system + systems[_target['SYSTEM']].send_system(_tmp_data) + #self._logger.debug('(%s) Packet routed by bridge: %s to system: %s TS: %s, TGID: %s', self._system, _bridge, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) # Final actions - Is this a voice terminator? - if (_frame_type == hb_const.HBPF_DATA_SYNC) and (_dtype_vseq == hb_const.HBPF_SLT_VTERM) and (self.STATUS[_slot]['RX_TYPE'] != hb_const.HBPF_SLT_VTERM): - call_duration = pkt_time - self.STATUS['RX_START'] + if (_frame_type == hb_const.HBPF_DATA_SYNC) and (_dtype_vseq == hb_const.HBPF_SLT_VTERM): + call_duration = pkt_time - self.STATUS['STREAM_START'] self._logger.info('(%s) *CALL END* STREAM ID: %s SUB: %s (%s) PEER: %s (%s) TGID %s (%s), TS %s, Duration: %s', \ self._system, int_id(_stream_id), get_alias(_rf_src, subscriber_ids), int_id(_rf_src), get_alias(_peer_id, peer_ids), int_id(_peer_id), get_alias(_dst_id, talkgroup_ids), int_id(_dst_id), _slot, call_duration) if CONFIG['REPORTS']['REPORT']: self._report.send_bridgeEvent('GROUP VOICE,END,{},{},{},{},{},{},{:.2f}'.format(self._system, int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _slot, int_id(_dst_id), call_duration)) - - - # Mark status variables for use later - self.STATUS[_slot]['RX_SEQ'] = _seq - self.STATUS[_slot]['RX_RFS'] = _rf_src - self.STATUS[_slot]['RX_TYPE'] = _dtype_vseq - self.STATUS[_slot]['RX_TGID'] = _dst_id - self.STATUS[_slot]['RX_TIME'] = pkt_time - self.STATUS[_slot]['RX_STREAM_ID'] = _stream_id - + removed = self.STATUS.pop(_stream_id) + if not removed: + self_logger.error('(%s) *CALL END* STREAM ID: %s NOT IN LIST -- THIS IS A REAL PROBLEM', self._system, int_id(_stream_id)) class routerHBP(HBSYSTEM): diff --git a/hblink.py b/hblink.py index 86aaef7..20a4adc 100755 --- a/hblink.py +++ b/hblink.py @@ -201,6 +201,7 @@ class OPENBRIDGE(DatagramProtocol): def send_system(self, _packet): if _packet[:4] == 'DMRD': _packet = _packet[:11] + self._config['NETWORK_ID'] + _packet[15:] + _packet += hmac_new(self._config['PASSPHRASE'],_packet,sha1).digest() self.transport.write(_packet, (self._config['TARGET_IP'], self._config['TARGET_PORT'])) # KEEP THE FOLLOWING COMMENTED OUT UNLESS YOU'RE DEBUGGING DEEPLY!!!! # self._logger.debug('(%s) TX Packet to OpenBridge %s:%s -- %s', self._system, self._config['TARGET_IP'], self._config['TARGET_PORT'], ahex(_packet)) @@ -216,7 +217,8 @@ class OPENBRIDGE(DatagramProtocol): # self._logger.debug('(%s) RX packet from %s -- %s', self._system, _sockaddr, ahex(_data)) if _packet[:4] == 'DMRD': # DMRData -- encapsulated DMR data frame - _data = _data[:53] + _data = _packet[:53] + _hash = _packet[53:] _ckhs = hmac_new(self._config['PASSPHRASE'],_data[53:],sha1).digest() if compare_digest(_hash, _ckhs) and _sockaddr == self._config['TARGET_SOCK']: