From 856b1dc2fd787138c623bb0db07b3eaa67259b21 Mon Sep 17 00:00:00 2001 From: IV3JDV Date: Sat, 8 Aug 2026 17:07:48 +0200 Subject: [PATCH] Fix routing RTP su NAT e supporto DTMF contemporaneo per piu utenti --- core.py | 80 ++++++++++++++++++++++++++++++++++++++++----------------- 1 file changed, 56 insertions(+), 24 deletions(-) diff --git a/core.py b/core.py index efedf8a..59853eb 100644 --- a/core.py +++ b/core.py @@ -31,7 +31,6 @@ 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) -# --- NUOVA VARIABILE PER LA SICUREZZA IP --- FREEPBX_IP = CONFIG.get("freepbx_ip", "127.0.0.1") def carica_rubrica(): @@ -44,7 +43,7 @@ def carica_rubrica(): UTENTI_SIP_DMR = carica_rubrica() # ==================================================================== -# CLASSE SESSIONE SIP (Con Buffer DTMF per il cambio TG) +# CLASSE SESSIONE SIP # ==================================================================== class SessioneSIP: def __init__(self, call_id, chiamante, destinatario_tg, rtp_addr, sip_addr, full_from, full_to, ip_vps): @@ -59,6 +58,9 @@ class SessioneSIP: 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 @@ -70,9 +72,10 @@ class SessioneSIP: self.ultimo_src_id = None - # --- VARIABILI PER IL CAMBIO TG TRAMITE DTMF --- + # --- 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 @@ -94,7 +97,6 @@ class OpenBridgeClient(asyncio.DatagramProtocol): src_id = int.from_bytes(data[5:8], 'big') tg_arrivo = str(int.from_bytes(data[8:11], 'big')) - # Il routing è dinamico: filtra in base al tg_destinazione attuale della sessione sessioni_interessate = [s for s in self.manager.sessioni.values() if s.tg_destinazione == tg_arrivo] if not sessioni_interessate: return @@ -194,10 +196,26 @@ class SessionManager: 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_client_addr and sess.rtp_client_addr[0] == addr[0]: - sess.rtp_client_addr = addr + 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): @@ -223,13 +241,12 @@ class SessionManager: sess.ptt_attivo = False if self.openbridge: self.openbridge.chiudi_flusso_chiamata(sess) - # --- LOGICA CAMBIO TG (DTMF 0-9) --- def gestisci_dtmf_digit(self, sess, digit): - if sess.ptt_attivo: return # Evita di cambiare TG mentre sei in TX + if sess.ptt_attivo: return + print(f"[DTMF] Utente {sess.chiamante} ha premuto: '{digit}'") sess.dtmf_buffer += str(digit) - # Resetta il task esistente se si preme un altro numero velocemente if sess.dtmf_task: sess.dtmf_task.cancel() @@ -237,21 +254,17 @@ class SessionManager: async def attendi_cambio_tg(self, sess): try: - # Attende 2.5 secondi dalla pressione dell'ultimo tasto await asyncio.sleep(2.5) if sess.dtmf_buffer: nuovo_tg = sess.dtmf_buffer - sess.dtmf_buffer = "" # Svuota il buffer + sess.dtmf_buffer = "" print(f"\n[Core] *** CAMBIO TG: {sess.chiamante} è passato al Talkgroup {nuovo_tg} ***") sess.tg_destinazione = nuovo_tg - - # Notifica il telefono del cambio avvenuto self.notifica_display(sess, f"TG: {nuovo_tg}") except asyncio.CancelledError: - pass # Il task è stato annullato perché l'utente ha premuto un altro tasto + pass - # --- FUNZIONE DISPLAY GENERICA (Ora accetta stringhe personalizzate) --- def notifica_display(self, sess, testo_display): if not self.sip_transport or not sess.sip_client_addr: return @@ -361,20 +374,22 @@ class RtpProtocol(asyncio.DatagramProtocol): 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: - evento_dtmf = data[12] - - # Meccanismo anti-rimbalzo (Debounce) per ignorare i frame duplicati ora = time.time() - if ora - sess.ultimo_tempo_rtp_dtmf < 0.2 and sess.ultimo_evento_dtmf == evento_dtmf: + 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 - # Esecuzione dei Comandi if evento_dtmf == 10: self.manager.gestisci_ptt(sess, True) elif evento_dtmf == 11: @@ -389,12 +404,8 @@ class SipProtocol(asyncio.DatagramProtocol): self.manager.sip_transport = transport def datagram_received(self, data, addr): - # --- SICUREZZA: BLOCCO IP --- if addr[0] != FREEPBX_IP and FREEPBX_IP != "0.0.0.0": - # Decommenta la riga sotto se vuoi vedere i log dei tentativi bloccati - # print(f"[SECURITY] Rifiutato pacchetto SIP da IP non autorizzato: {addr[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} @@ -436,6 +447,27 @@ class SipProtocol(asyncio.DatagramProtocol): 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]