import json from bitstring import BitArray import asyncio import struct import time import os import audioop from vocoder_engine import VocoderEngine from openbridge_protocol import crea_pacchetto_ping, calcola_hmac from dmr_utils3.ambe_utils import convert49BitTo72BitAMBE, convert72BitTo49BitAMBE # ==================================================================== # CARICAMENTO CONFIGURAZIONI (JSON) # ==================================================================== def carica_config(): try: with open("config.json", "r") as f: return json.load(f) except Exception as e: print(f"[ERRORE CRITICO] File config.json non trovato o errato ({e}). Chiusura.") exit(1) CONFIG = carica_config() HBLINK_IP = CONFIG.get("hblink_ip", "127.0.0.1") HBLINK_PORT = CONFIG.get("hblink_port", 62047) HBLINK_PASSWORD = CONFIG.get("hblink_password", "TestVOIP") PEER_ID = CONFIG.get("peer_id", 22241) SIP_PORT = CONFIG.get("sip_port", 55060) RTP_PORT = CONFIG.get("rtp_port", 10000) VOCODER_HOST = CONFIG.get("vocoder_host", "127.0.0.1") VOCODER_PORT = CONFIG.get("vocoder_port", 2470) FREEPBX_IP = CONFIG.get("freepbx_ip", "127.0.0.1") def carica_rubrica(): try: with open("utenti.json", "r") as f: return json.load(f) except Exception as e: return {} UTENTI_SIP_DMR = carica_rubrica() # ==================================================================== # CLASSE SESSIONE SIP # ==================================================================== class SessioneSIP: def __init__(self, call_id, chiamante, destinatario_tg, rtp_addr, sip_addr, full_from, full_to, ip_vps): self.call_id = call_id self.chiamante = chiamante self.tg_destinazione = str(destinatario_tg) self.dmr_id = int(UTENTI_SIP_DMR.get(str(chiamante), 2223619)) self.rtp_client_addr = rtp_addr self.sip_client_addr = sip_addr self.ip_vps = ip_vps self.rtp_seq = 0 self.rtp_timestamp = 0 # Flag per la gestione NAT delle porte RTP self.rtp_confermato = False self.dialog_to = full_from self.dialog_from = full_to.split(';')[0] + ";tag=987654" self.local_cseq = 10 self.ptt_attivo = False self.buffer_ambe = [] self.seq_dmr = 0 self.stream_id = os.urandom(4) self.ultimo_src_id = None # --- VARIABILI DTMF --- self.dtmf_buffer = "" self.dtmf_task = None self.ultimo_rtp_timestamp_dtmf = None self.ultimo_tempo_rtp_dtmf = 0 self.ultimo_evento_dtmf = None class OpenBridgeClient(asyncio.DatagramProtocol): def __init__(self, manager): self.transport = None self.manager = manager def connection_made(self, transport): self.transport = transport print(f"[OpenBridge] Socket connesso a {HBLINK_IP}:{HBLINK_PORT}") asyncio.create_task(self.keepalive_loop()) def datagram_received(self, data, addr): if len(data) < 53: return if data.startswith(b"DMRD") and len(data) >= 73: src_id = int.from_bytes(data[5:8], 'big') tg_arrivo = str(int.from_bytes(data[8:11], 'big')) sessioni_interessate = [s for s in self.manager.sessioni.values() if s.tg_destinazione == tg_arrivo] if not sessioni_interessate: return print("D", end="", flush=True) payload_33 = data[20:53] ambe1 = payload_33[0:9] ambe2 = bytearray(9) ambe2[0:4] = payload_33[9:13] ambe2[4] = (payload_33[13] & 0xF0) | (payload_33[19] & 0x0F) ambe2[5:9] = payload_33[20:24] ambe3 = payload_33[24:33] frame_uniti = ambe1 + bytes(ambe2) + ambe3 self.manager.ricevi_frame_dmr_multi(frame_uniti, sessioni_interessate, src_id) async def keepalive_loop(self): while True: if self.transport: pacchetto_ping = crea_pacchetto_ping(PEER_ID, HBLINK_PASSWORD) self.transport.sendto(pacchetto_ping) await asyncio.sleep(10) def apri_flusso_chiamata(self, sess): if not self.transport: return flags = 0 payload_base = struct.pack("!3sB I 4s B I I", b"OPB", 0x80, PEER_ID, sess.stream_id, flags, sess.dmr_id, int(sess.tg_destinazione)) firma_hmac = calcola_hmac(HBLINK_PASSWORD, payload_base) self.transport.sendto(payload_base + firma_hmac) def invia_dmrd_voce(self, sess, payload_ambe): if not self.transport: return try: magic = b"DMRD" seq_byte = struct.pack("B", sess.seq_dmr) src_bytes = sess.dmr_id.to_bytes(3, 'big') dst_bytes = int(sess.tg_destinazione).to_bytes(3, 'big') peer_bytes = PEER_ID.to_bytes(4, 'big') flags_byte = struct.pack("B", 0x00) header = magic + seq_byte + src_bytes + dst_bytes + peer_bytes + flags_byte + sess.stream_id payload_ambe = payload_ambe.ljust(27, b'\x00') ambe1 = payload_ambe[0:9] ambe2 = payload_ambe[9:18] ambe3 = payload_ambe[18:27] payload_33 = bytearray(33) payload_33[0:9] = ambe1 payload_33[9:13] = ambe2[0:4] payload_33[13] = ambe2[4] & 0xF0 payload_33[19] = ambe2[4] & 0x0F payload_33[20:24] = ambe2[5:9] payload_33[24:33] = ambe3 pacchetto_base = header + bytes(payload_33) firma_hmac = calcola_hmac(HBLINK_PASSWORD, pacchetto_base) self.transport.sendto(pacchetto_base + firma_hmac) except Exception: pass def chiudi_flusso_chiamata(self, sess): if not self.transport: return magic = b"DMRD" flags_byte = struct.pack("B", 0x02) src_bytes = sess.dmr_id.to_bytes(3, 'big') dst_bytes = int(sess.tg_destinazione).to_bytes(3, 'big') peer_bytes = PEER_ID.to_bytes(4, 'big') for i in range(3): seq_byte = struct.pack("B", (sess.seq_dmr + i) % 256) header = magic + seq_byte + src_bytes + dst_bytes + peer_bytes + flags_byte + sess.stream_id pacchetto_base = header + (b'\x00' * 33) firma_hmac = calcola_hmac(HBLINK_PASSWORD, pacchetto_base) self.transport.sendto(pacchetto_base + firma_hmac) # ==================================================================== # GESTORE CENTRALE DELLE SESSIONI # ==================================================================== class SessionManager: def __init__(self): self.vocoder = VocoderEngine(mode="emulator", host=VOCODER_HOST, port=VOCODER_PORT) self.openbridge = None self.rtp_transport = None self.sip_transport = None self.sessioni = {} async def avvia(self): await self.vocoder.avvia() loop = asyncio.get_running_loop() _, self.openbridge = await loop.create_datagram_endpoint( lambda: OpenBridgeClient(self), local_addr=('0.0.0.0', HBLINK_PORT), remote_addr=(HBLINK_IP, HBLINK_PORT) ) print(f"[Core] PBX Multi-Utente avviato. SIP su porta {SIP_PORT}...\n") print(f"[Security] Accetto connessioni SIP SOLO da: {FREEPBX_IP}\n") def trova_sessione_per_rtp(self, addr): # 1. Match tra le sessioni già confermate (IP + Porta NAT reale) for sess in self.sessioni.values(): if sess.rtp_confermato and sess.rtp_client_addr == addr: return sess # 2. Match esatto per sessioni non ancora confermate (es. senza NAT) for sess in self.sessioni.values(): if not sess.rtp_confermato and sess.rtp_client_addr == addr: sess.rtp_confermato = True return sess # 3. Gestione NAT: Assegna la porta NAT alla prima sessione ancora in attesa per quell'IP non_confermate = [s for s in self.sessioni.values() if s.rtp_client_addr and s.rtp_client_addr[0] == addr[0] and not s.rtp_confermato] if non_confermate: sess = non_confermate[0] sess.rtp_client_addr = addr sess.rtp_confermato = True print(f"[RTP] Flusso Audio/DTMF agganciato per l'utente {sess.chiamante} ({addr[0]}:{addr[1]})") return sess return None def gestisci_nuova_chiamata(self, call_id, chiamante, destinatario, rtp_addr, sip_addr, full_from, full_to, ip_vps): sess = SessioneSIP(call_id, chiamante, destinatario, rtp_addr, sip_addr, full_from, full_to, ip_vps) self.sessioni[call_id] = sess print(f"\n[Core] >>> SIP {chiamante} -> TG {destinatario} | ID: {sess.dmr_id}") def termina_chiamata(self, call_id): if call_id in self.sessioni: sess = self.sessioni[call_id] if sess.ptt_attivo and self.openbridge: self.openbridge.chiudi_flusso_chiamata(sess) del self.sessioni[call_id] def gestisci_ptt(self, sess, stato): if stato is True and not sess.ptt_attivo: sess.ptt_attivo = True sess.buffer_ambe = [] sess.seq_dmr = 0 sess.stream_id = os.urandom(4) if self.openbridge: self.openbridge.apri_flusso_chiamata(sess) elif stato is False and sess.ptt_attivo: sess.ptt_attivo = False if self.openbridge: self.openbridge.chiudi_flusso_chiamata(sess) def gestisci_dtmf_digit(self, sess, digit): if sess.ptt_attivo: return print(f"[DTMF] Utente {sess.chiamante} ha premuto: '{digit}'") sess.dtmf_buffer += str(digit) if sess.dtmf_task: sess.dtmf_task.cancel() sess.dtmf_task = asyncio.create_task(self.attendi_cambio_tg(sess)) async def attendi_cambio_tg(self, sess): try: await asyncio.sleep(2.5) if sess.dtmf_buffer: nuovo_tg = sess.dtmf_buffer sess.dtmf_buffer = "" print(f"\n[Core] *** CAMBIO TG: {sess.chiamante} è passato al Talkgroup {nuovo_tg} ***") sess.tg_destinazione = nuovo_tg self.notifica_display(sess, f"TG: {nuovo_tg}") except asyncio.CancelledError: pass def notifica_display(self, sess, testo_display): if not self.sip_transport or not sess.sip_client_addr: return sess.local_cseq += 1 branch = f"z9hG4bK-{os.urandom(4).hex()}" client_sip_ip, client_sip_port = sess.sip_client_addr sdp = (f"v=0\r\no=- 12345 12345 IN IP4 {sess.ip_vps}\r\ns=Talk\r\n" f"c=IN IP4 {sess.ip_vps}\r\nt=0 0\r\nm=audio {RTP_PORT} RTP/AVP 8 101\r\n" f"a=rtpmap:8 PCMA/8000\r\na=rtpmap:101 telephone-event/8000\r\na=sendrecv\r\n") sip_reinvite = ( f"INVITE sip:{sess.chiamante}@{client_sip_ip}:{client_sip_port} SIP/2.0\r\n" f"Via: SIP/2.0/UDP 127.0.0.1:{SIP_PORT};branch={branch}\r\n" f"Max-Forwards: 70\r\n" f"From: {sess.dialog_from}\r\n" f"To: {sess.dialog_to}\r\n" f"Call-ID: {sess.call_id}\r\n" f"CSeq: {sess.local_cseq} INVITE\r\n" f"Contact: \r\n" f"Remote-Party-ID: \"{testo_display}\" ;party=called;screen=yes;privacy=off\r\n" f"P-Asserted-Identity: \"{testo_display}\" \r\n" f"Content-Type: application/sdp\r\n" f"Content-Length: {len(sdp)}\r\n" f"\r\n" f"{sdp}" ) self.sip_transport.sendto(sip_reinvite.encode('utf-8'), sess.sip_client_addr) def ricevi_frame_dmr_multi(self, ambe_frames, sessioni_interessate, src_id): ascoltatori = [s for s in sessioni_interessate if not s.ptt_attivo] if not ascoltatori or not self.rtp_transport: return for sess in ascoltatori: if sess.ultimo_src_id != src_id: sess.ultimo_src_id = src_id self.notifica_display(sess, f"RX: {src_id}") asyncio.create_task(self.processa_audio_dmr(ambe_frames, ascoltatori)) async def processa_audio_dmr(self, ambe_27_bytes, ascoltatori): try: for i in range(0, len(ambe_27_bytes), 9): frame_9 = ambe_27_bytes[i:i+9] if len(frame_9) < 9: continue try: ambe_49_bits = convert72BitTo49BitAMBE(BitArray(bytes=frame_9)) ba_49 = ambe_49_bits if isinstance(ambe_49_bits, BitArray) else BitArray(ambe_49_bits) if len(ba_49) < 56: ba_49.append(BitArray(uint=0, length=56 - len(ba_49))) frame_7 = ba_49.bytes[:7] except Exception: continue pcm_data = await self.vocoder.converti_ambe_in_pcm(frame_7) if not pcm_data or len(pcm_data) < 2: continue if len(pcm_data) % 2 != 0: pcm_data = pcm_data[:-1] pcma_data = audioop.lin2alaw(pcm_data, 2) for sess in ascoltatori: header = bytearray(12) header[0] = 0x80; header[1] = 0x08 header[2:4] = struct.pack("!H", sess.rtp_seq) header[4:8] = struct.pack("!I", sess.rtp_timestamp) header[8:12] = struct.pack("!I", 0x12345678) rtp_packet = header + pcma_data if sess.rtp_client_addr: self.rtp_transport.sendto(rtp_packet, sess.rtp_client_addr) sess.rtp_seq = (sess.rtp_seq + 1) % 65536 sess.rtp_timestamp = (sess.rtp_timestamp + 160) % 4294967296 except Exception: pass async def processa_audio_rtp(self, sess, rtp_payload): if not sess.ptt_attivo: return try: ambe_bytes = await asyncio.wait_for(self.vocoder.converti_rtp_in_ambe(rtp_payload), timeout=0.1) if not ambe_bytes: return dim_frame = 7 if len(ambe_bytes) % 7 == 0 else 9 for i in range(0, len(ambe_bytes), dim_frame): frame = ambe_bytes[i:i+dim_frame] if len(frame) == 7: stringa_bit = BitArray(bytes=frame).bin ambe_49_bits = [int(b) for b in stringa_bit[:49]] fec_9_byte = convert49BitTo72BitAMBE(ambe_49_bits) sess.buffer_ambe.append(bytes(fec_9_byte)) elif len(frame) == 9: sess.buffer_ambe.append(frame) while len(sess.buffer_ambe) >= 3: payload_vocale = b''.join(sess.buffer_ambe[:3]) sess.buffer_ambe = sess.buffer_ambe[3:] self.openbridge.invia_dmrd_voce(sess, payload_vocale) sess.seq_dmr = (sess.seq_dmr + 1) % 256 except Exception: pass # ==================================================================== # PROTOCOLLI DI RETE SIP/RTP # ==================================================================== class RtpProtocol(asyncio.DatagramProtocol): def __init__(self, manager): self.manager = manager def connection_made(self, transport): self.manager.rtp_transport = transport def datagram_received(self, data, addr): sess = self.manager.trova_sessione_per_rtp(addr) if not sess or len(data) < 12: return payload_type = data[1] & 0x7F if payload_type == 8: asyncio.create_task(self.manager.processa_audio_rtp(sess, data[12:])) elif payload_type == 101: ora = time.time() rtp_timestamp = struct.unpack("!I", data[4:8])[0] evento_dtmf = data[12] if rtp_timestamp == sess.ultimo_rtp_timestamp_dtmf or (ora - sess.ultimo_tempo_rtp_dtmf < 0.20 and sess.ultimo_evento_dtmf == evento_dtmf): return sess.ultimo_rtp_timestamp_dtmf = rtp_timestamp sess.ultimo_tempo_rtp_dtmf = ora sess.ultimo_evento_dtmf = evento_dtmf if evento_dtmf == 10: self.manager.gestisci_ptt(sess, True) elif evento_dtmf == 11: self.manager.gestisci_ptt(sess, False) elif 0 <= evento_dtmf <= 9: self.manager.gestisci_dtmf_digit(sess, evento_dtmf) class SipProtocol(asyncio.DatagramProtocol): def __init__(self, manager): self.manager = manager def connection_made(self, transport): self.transport = transport self.manager.sip_transport = transport def datagram_received(self, data, addr): if addr[0] != FREEPBX_IP and FREEPBX_IP != "0.0.0.0": return msg = data.decode('utf-8', errors='ignore') headers = {line.split(":", 1)[0].strip().upper(): line.split(":", 1)[1].strip() for line in msg.split('\r\n')[1:] if ":" in line} call_id = headers.get('CALL-ID', '') if msg.startswith("OPTIONS"): resp = (f"SIP/2.0 200 OK\r\nVia: {headers.get('VIA', '')}\r\n" f"From: {headers.get('FROM', '')}\r\nTo: {headers.get('TO', '')}\r\n" f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\n" f"Contact: \r\n" f"Content-Length: 0\r\n\r\n") self.transport.sendto(resp.encode('utf-8'), addr) elif msg.startswith("INVITE"): client_rtp_port = 4000 for line in msg.split('\n'): line = line.strip() if line.startswith("m=audio "): client_rtp_port = int(line.split()[1]) ip_vps = "127.0.0.1" if "@" in headers.get("TO", ""): ip_vps = headers.get("TO", "").split("@")[1].split(">")[0] chiamante = headers.get("FROM", "").split("sip:")[1].split("@")[0] destinatario = headers.get("TO", "").split("sip:")[1].split("@")[0] full_from = headers.get('FROM', '') full_to = headers.get('TO', '') self.manager.gestisci_nuova_chiamata(call_id, chiamante, destinatario, (addr[0], client_rtp_port), addr, full_from, full_to, ip_vps) sdp = (f"v=0\r\no=- 12345 12345 IN IP4 {ip_vps}\r\ns=Talk\r\n" f"c=IN IP4 {ip_vps}\r\nt=0 0\r\nm=audio {RTP_PORT} RTP/AVP 8 101\r\n" f"a=rtpmap:8 PCMA/8000\r\na=rtpmap:101 telephone-event/8000\r\na=sendrecv\r\n") resp = (f"SIP/2.0 200 OK\r\nVia: {headers.get('VIA', '')}\r\n" f"From: {full_from}\r\nTo: {full_to};tag=987654\r\n" f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\n" f"Contact: \r\nContent-Type: application/sdp\r\n" f"Content-Length: {len(sdp)}\r\n\r\n{sdp}") self.transport.sendto(resp.encode('utf-8'), addr) elif msg.startswith("INFO"): resp = (f"SIP/2.0 200 OK\r\nVia: {headers.get('VIA', '')}\r\n" f"From: {headers.get('FROM', '')}\r\nTo: {headers.get('TO', '')}\r\n" f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\n" f"Content-Length: 0\r\n\r\n") self.transport.sendto(resp.encode('utf-8'), addr) sess = self.manager.sessioni.get(call_id) if sess: for line in msg.split('\r\n'): if line.upper().startswith("SIGNAL="): signal = line.split("=")[1].strip() if signal == "*": evento = 10 elif signal == "#": evento = 11 elif signal.isdigit(): evento = int(signal) else: continue if evento == 10: self.manager.gestisci_ptt(sess, True) elif evento == 11: self.manager.gestisci_ptt(sess, False) elif 0 <= evento <= 9: self.manager.gestisci_dtmf_digit(sess, evento) elif msg.startswith("SIP/2.0 200 OK"): if "INVITE" in headers.get('CSEQ', ''): cseq_num = headers.get('CSEQ', '').split()[0] ack_uri = f"sip:{addr[0]}:{addr[1]}" if "CONTACT" in headers: contact_hdr = headers["CONTACT"] if "<" in contact_hdr: ack_uri = contact_hdr.split("<")[1].split(">")[0] ack = (f"ACK {ack_uri} SIP/2.0\r\n" f"Via: SIP/2.0/UDP 127.0.0.1:{SIP_PORT};branch=z9hG4bK-{os.urandom(4).hex()}\r\n" f"Max-Forwards: 70\r\n" f"From: {headers.get('FROM', '')}\r\n" f"To: {headers.get('TO', '')}\r\n" f"Call-ID: {call_id}\r\n" f"CSeq: {cseq_num} ACK\r\n" f"Content-Length: 0\r\n\r\n") self.transport.sendto(ack.encode('utf-8'), addr) elif msg.startswith("BYE"): self.manager.termina_chiamata(call_id) resp = (f"SIP/2.0 200 OK\r\nVia: {headers.get('VIA', '')}\r\n" f"From: {headers.get('FROM', '')}\r\nTo: {headers.get('TO', '')}\r\n" f"Call-ID: {call_id}\r\nCSeq: {headers.get('CSEQ', '')}\r\n" f"Content-Length: 0\r\n\r\n") self.transport.sendto(resp.encode('utf-8'), addr) async def main(): loop = asyncio.get_running_loop() manager = SessionManager() await manager.avvia() _, _ = await loop.create_datagram_endpoint(lambda: SipProtocol(manager), local_addr=('0.0.0.0', SIP_PORT)) _, _ = await loop.create_datagram_endpoint(lambda: RtpProtocol(manager), local_addr=('0.0.0.0', RTP_PORT)) try: await asyncio.Event().wait() except KeyboardInterrupt: pass if __name__ == "__main__": asyncio.run(main())