#!/usr/bin/python3 # pYSFReflector # # Created by Antonio Matraia (IU5JAE) on 20/02/2021. # Copyright 2021 Antonio Matraia (IU5JAE). All rights reserved. # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . import socket import threading import queue import sys import os import time import re import configparser import datetime import signal from datetime import datetime import bisect import struct def ip2long(ip): packed = socket.inet_aton(ip) lng = struct.unpack("!L", packed)[0] return lng def long2ip(lng): packed = struct.pack("!L", lng) ip = socket.inet_ntoa(packed) return ip ## gestione liste LH ecc ## def inserisci_lista(lista, elemento, n_max): if (len(lista) < n_max): lista.append(elemento) else: for i in range(n_max - 1): lista[i] = lista[i+1] lista[n_max-1] = elemento def stampa_lista(lista): for i in range(len(lista)): print(lista[len(lista)-i-1]) ################################# def inlist(a, x): i = bisect.bisect_left(a, x) if i != len(a) and a[i] == x: return True else: return False def check_string(l): s = '' for c in l: if ((not c.isprintable()) and (ord(c) != 10)): s += '<' +str(ord(c)) + '>' if c.isprintable(): s += c return s def RecvData(sock,recvPackets): while True: data,addr = sock.recvfrom(1024) # bloccante se non ricevo niente recvPackets.put((data,addr)) def CalcID(ref): c = ref.strip().ljust(16) u = 0 for a in c: u = (u + ord(a)) & 0xFFFFFFFF u = (u + (u << 10) & 0xFFFFFFFF) & 0xFFFFFFFF u = (u ^ (u >> 6)) & 0xFFFFFFFF u = (u + (u << 3) & 0xFFFFFFFF) & 0xFFFFFFFF u = (u ^ (u >> 11))& 0xFFFFFFFF u = (u + (u << 15)& 0xFFFFFFFF)& 0xFFFFFFFF u = u % 100000 return u def getidgw(cl, adr): i=[0, ''] for c in cl: if ((c[0] == adr[0]) and (c[1] == adr[1])): i = [c[4], c[2]] break return i def ElencoNodi(cl): while True: time.sleep(120) cl_lo = [] if (len(cl) == 0): printlog('No repeaters/gateways linked') else: printlog('Currently linked repeaters/gateways:') for c in cl: printlog(' ' + c[2].ljust(10) + ': ' + str(c[0]) + ':' + str(c[1]) + ' ' + str(c[3]) + '/60') if (c[5] == 1): cl_lo.append(c) if (len(cl_lo) == 0): printlog('No repeaters/gateways muted') else: printlog('Currently muted repeaters/gateways:') for c in cl_lo: printlog(' ' + c[2].ljust(10) + ': ' + str(c[0]) + ':' + str(c[1]) + ' ' + str(c[3]) + '/60') def TimeoutNodi(cl): while True: time.sleep(1) for c in cl: c[3] += 1 if (c[3] > 60): printlog('Removing ' + c[2].ljust(10) + ' (' + c[0] + ':' + str(c[1]) + ') disappeared') cl.remove(c) def TimeoutTX(t, t_lock, r_lock, lista_lh): while True: if (t[1] < 5): t[1] += 0.1 if ((t[1] > 1.0) and (t[0] != 0)): t[0] = 0 printlog('Network watchdog has expired') inserisci_lista(lista_lh, [check_string(t[2]), check_string(t[3]), check_string(t[4]), t[5], datetime.fromtimestamp(t[6]).strftime("%d-%m-%Y %H-%M-%S"), round(time.time() - t[6]) ], 20) t[0] = 0 t[2] = '' t[3] = '' t[4] = '' t[5] = 0 t[6] = 0 pop_list = [] for d in t_lock: if (t_lock[d] < 5): t_lock[d] += 0.1 if ((t_lock[d] > 1.5) and (t_lock[d] != 0)): pop_list.append(d) r_lock.remove(d) for x in pop_list: t_lock.pop(x) printlog('Removed from blockeds queue ' + str(x)) time.sleep(0.1) def canTrasmit(cs, re_en): global BLACK_LIST global WHITE_LIST call_sp = re.split(r'[-/]', cs) call = call_sp[0] if inlist(WHITE_LIST, call): return True if inlist(BLACK_LIST, call): return False if (len(call_sp) > 1): if (call_sp[1] == 'RPT'): return False if re_en: if (re.match(r'^\d?[A-Z]{1,2}\d{1,4}[A-Z]{1,3}$',call,re.IGNORECASE) and (len(call) <= 8)): return True else: return False else: return True def lista_gw(cl): info = '' for c in cl: info += c[2] + ':' + c[0] + ':' + str(c[1]) + ';' return info def lista_invio(lista): info = '' for i in range(len(lista)): info += lista[len(lista)-i-1][0] + ':' + lista[len(lista)-i-1][1] + ':' + lista[len(lista)-i-1][2] + ':' + str(lista[len(lista)-i-1][3]) + ':' + lista[len(lista)-i-1][4] + ':' + str(lista[len(lista)-i-1][5]) + ';' return info def update_clients(cl): global GW_BL global IP_BL for c in cl: if (inlist(GW_BL, c[2]) or inlist(IP_BL, ip2long(c[0]))): c[5] = 1 else: c[5] = 0 def blacklist(f_bl, t_reload, cli): global BLACK_LIST global WHITE_LIST global GW_BL global IP_BL f_time_old = 0 try: f_time = os.stat(f_bl).st_mtime except: pass while True: f_time = os.stat(f_bl).st_mtime if (f_time != f_time_old): try: file = open(f_bl) BL_TMP = [] GW_TMP = [] IP_TMP = [] WL_TMP = [] printlog('Reload the Blacklist from File') for row in file: content = row.strip() # riga valida if ((len(content) > 3) and (content[0] != '#')): c_split = content.split(':') # CALL if (len(c_split) == 1 or c_split[0] == 'CS'): if (len(c_split) == 1): cont = content if (c_split[0] == 'CS'): cont = c_split[1] if ((len(cont) <= 8) and (len(cont) >= 3)): if (not inlist(BL_TMP, cont)): bisect.insort(BL_TMP,cont) # WL if (len(c_split) == 2 and c_split[0] == 'AL'): if (not inlist(WL_TMP, c_split[1])): bisect.insort(WL_TMP,c_split[1]) # GW if (len(c_split) == 2 and c_split[0] == 'GW'): if (not inlist(GW_TMP, c_split[1])): bisect.insort(GW_TMP,c_split[1]) # IP if (len(c_split) == 2 and c_split[0] == 'IP'): try: ipa = socket.gethostbyname(c_split[1]) ipl = ip2long(ipa) except: ipl = 0 printlog('Invalid hostname ' + c_split[1]) if (ipl > 0): if (not inlist(IP_TMP, ipl)): bisect.insort(IP_TMP, ipl) file.close() except Exception as ex: printlog('Failed to load Blacklist from File ') BLACK_LIST = BL_TMP.copy() WHITE_LIST = WL_TMP.copy() GW_BL = GW_TMP.copy() IP_BL = IP_TMP.copy() f_time_old = f_time update_clients(cli) else: pass time.sleep(t_reload) ## lettura file configurazione ## def ReadConfig(f,p): config = configparser.ConfigParser() config_file = f.strip() config.read(config_file) name = config['Info']['Name'] description = config['Info']['Description'] try: id = int(config['Info']['id']) except: id = 0 log_path = config['Log']['FilePath'] log_name = config['Log']['FileRoot'] try: file_rotate = config['Log']['FileRotate'] except: file_rotate = "1" # to keep file-rotation by default with timestamp try: port = int(config['Network']['Port']) except: port = 42000 try: file_blacklist = config['Block List']['File'] except: file_blacklist = '' try: CheckRE = int(config['Block List']['CheckRE']) except: CheckRE = 1 try: t_reload_blacklist = float(config['Block List']['Time']) except: t_reload_blacklist = 5.0 if (t_reload_blacklist < 0.1): t_reload_blacklist = 0.1 p.append(id) # 0 p.append(name) # 1 p.append(description) # 2 p.append(log_path) # 3 p.append(log_name) # 4 p.append(port) # 5 p.append(file_blacklist) # 6 p.append(t_reload_blacklist) # 7 p.append(file_rotate) # 8 p.append(CheckRE) # 9 def sanitize_msg(data): bya_msg = bytearray(data) if ((data[0:4] == b"YSFP") and (len(data) == 14)): for i in range(10): if (bya_msg[i+4] == 0): bya_msg[i+4] = 32 if ((data[0:4] == b"YSFD") and (len(data) == 155)): for i in range(30): if (bya_msg[i+4] == 0): bya_msg[i+4] = 32 return(bytes(bya_msg)) def RunServer(config): global filelog global version global BLACK_LIST global GW_BL global IP_BL host = '0.0.0.0' port = config[5] s = socket.socket(socket.AF_INET,socket.SOCK_DGRAM) s.setblocking(1) s.bind((host,port)) clients = [] # addr, port, gw, t_corr, ID, lonly c = [] rx_lock = [] rx_lock_tout = {} # lista LH LH = [] # lista bloccati BL = [] id = 1 # id nodo id_str = 1 # id stream tx = [0, 0, '', '', '', 0, 0] # id_nodo, tout, gateway, src, dest, id_stream, start_time refl_name = config[1] refl_desc = config[2] if ((config[0] > 0) and (config[0] < 1000000)): refl_id = str(config[0]).zfill(5) else: refl_id = CalcID(refl_name) f_blacklist = config[6] tr_blacklist = config[7] * 60.0 CheckRE = config[9] recvPackets = queue.Queue() print('Starting YSFReflector-' + version) printlog('Starting YSFReflector-' + version) threading.Thread(target=RecvData,args=(s,recvPackets)).start() threading.Thread(target=ElencoNodi,args=(clients,)).start() threading.Thread(target=TimeoutNodi,args=(clients,)).start() threading.Thread(target=TimeoutTX,args=(tx,rx_lock_tout,rx_lock,LH)).start() if (len(f_blacklist) > 0): threading.Thread(target=blacklist,args=(f_blacklist,tr_blacklist,clients)).start() time_start = time.time() while True: data_ns, addr = recvPackets.get() # bloccante se coda vuota data = sanitize_msg(data_ns) cmd = data[0:4] if (cmd == b'YSFP'): pres = False for c in clients: if ((c[0] == addr[0]) and (c[1] == addr[1])): pres = True c[3] = 0 break if not pres: lonly = 0 if inlist(GW_BL, (data[4:14]).decode().strip()): lonly = 1 if inlist(IP_BL, ip2long(addr[0])): lonly = 1 c=[addr[0], addr[1], (data[4:14]).decode().strip(), 0, id, lonly] id += 1 clients.append(c) printlog('Adding ' + c[2].ljust(10) + ' (' + c[0] + ':' + str(c[1]) + ')') s.sendto(b'YSFPREFLECTOR ',addr) if (cmd == b'YSFU'): for c in clients: if ((c[0] == addr[0]) and (c[1] == addr[1])): printlog('Removing ' + c[2].ljust(10) + ' (' + c[0] + ':' + str(c[1]) + ') unlinked') clients.remove(c) break if ((cmd == b'YSFD') and (len(data) == 155)): [id_corr, gw_corr] = getidgw(clients, addr) if (tx[0] == 0): if inlist(GW_BL, gw_corr): tx_ok = False block_r = 'GW' else: if inlist(IP_BL, ip2long(addr[0])): tx_ok = False block_r = 'IP' else: tx_ok = canTrasmit(data[14:24].decode().strip(), CheckRE) block_r = 'CS' if tx_ok: if ((tx[0] == 0) and (id_corr != 0)): tx[0] = id_corr # gateway tx[2] = data[4:14].decode().strip() # src tx[3] = data[14:24].decode().strip() # dest tx[4] = data[24:34].decode().strip() # stream ID tx[5] = id_str # time start tx[6] = time.time() id_str += 1 printlog('Received data from ' + tx[3].ljust(10) + ' to ' + tx[4].ljust(10) + ' at ' + tx[2].ljust(10)) else: if (id_corr not in rx_lock): rx_lock.append(id_corr) inserisci_lista(BL, [check_string(data[4:14].decode().strip()), check_string(data[14:24].decode().strip()), check_string(data[24:34].decode().strip()), -1, datetime.fromtimestamp(time.time()).strftime("%d-%m-%Y %H-%M-%S"), -1 ], 20) printlog('Data from ' + data[14:24].decode().strip().ljust(10) + ' at ' + data[4:14].decode().strip().ljust(10) + ' blocked/' + block_r) # printlog('Bloccato: ' + data[14:24].decode().strip() + ' via ' + data[4:14].decode().strip()) rx_lock_tout[id_corr] = 0 if ((id_corr == tx[0]) and (id_corr != 0)): tx[1] = 0 for c in clients: if (((c[0] != addr[0]) or (c[1] != addr[1])) and (id_corr == tx[0]) and (id_corr != 0) and (id_corr not in rx_lock)): s.sendto(data,(c[0], c[1])) if ((data[34] & 0x01) == 0x01): if (tx[0] != 0): printlog('Received end of transmission') inserisci_lista(LH, [check_string(tx[2]), check_string(tx[3]), check_string(tx[4]), tx[5], datetime.fromtimestamp(tx[6]).strftime("%d-%m-%Y %H-%M-%S"), round(time.time() - tx[6]) ], 20) tx[0] = 0 tx[2] = '' tx[3] = '' tx[4] = '' tx[5] = 0 tx[6] = 0 if (cmd == b'YSFS'): # printlog('YSF server status enquiry from ' + addr[0] + ':' + str(addr[1])) if (len(clients) > 999): num_cli = 999 else: num_cli = len(clients) info = 'YSFS' + str(refl_id).zfill(5) + refl_name.ljust(16) + refl_desc.ljust(14) + str(num_cli).zfill(3) s.sendto(str.encode(info),addr) ## messaggi per report attivo ## if (cmd == b'QSRU'): printlog('Received command ' + cmd.decode() + ' from: ' + addr[0] + ':' + str(addr[1])) info = 'ASRU;' + str(round(time.time()-time_start)) + ';' s.sendto(str.encode(info),addr) if (cmd == b'QSRI'): printlog('Received command ' + cmd.decode() + ' from: ' + addr[0] + ':' + str(addr[1])) info = 'ASRI;' + str(refl_id) + ':' + refl_name + ':' + refl_desc + ';' s.sendto(str.encode(info),addr) if (cmd == b'QGWL'): printlog('Received command ' + cmd.decode() + ' from: ' + addr[0] + ':' + str(addr[1])) info = 'AGWL;' + lista_gw(clients) s.sendto(str.encode(info),addr) if (cmd == b'QLHL'): printlog('Received command ' + cmd.decode() + ' from: ' + addr[0] + ':' + str(addr[1])) info = 'ALHL;' + lista_invio(LH) s.sendto(str.encode(info),addr) if (cmd == b'QREJ'): printlog('Received command ' + cmd.decode() + ' from: ' + addr[0] + ':' + str(addr[1])) info = 'AREJ;' + lista_invio(BL) s.sendto(str.encode(info),addr) if (cmd == b'QBLK'): printlog('Received command ' + cmd.decode() + ' from: ' + addr[0] + ':' + str(addr[1])) s_info = '' for c in BLACK_LIST: s_info += c + ';' info = 'ABLK;' + s_info s.sendto(str.encode(info),addr) s.close() def printlog(mess): global filelog global log_basename global file_rotate if file_rotate == "1": log_file = log_basename + '-' + str(datetime.utcnow().strftime('%Y-%m-%d')) + '.log' else: log_file = log_basename + '.log' try: if not os.path.isfile(log_file): filelog.flush() filelog.close() filelog = open(log_file,'x') except: pass str_log = check_string('M: ' + str(datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S.%f'))[:-3] + ' ' + mess) try: filelog.write(str_log + '\n') filelog.flush() except: pass ######## main ######## version = '20210322' if (len(sys.argv) != 2): print('Invalid Number of Arguments') print('use: YSFReflector ') sys.exit() if (sys.argv[1].strip() == '-v'): print('YSFReflector version ' + version) sys.exit() ## lettura configurazione ## config=[] try: ReadConfig(sys.argv[1].strip(), config) except: print('Unable to read configuration file') sys.exit() log_basename = config[3] + '/' + config[4] file_rotate = config[8] ### log if file_rotate == "1": log_file = log_basename + '-' + str(datetime.utcnow().strftime('%Y-%m-%d')) + '.log' else: log_file = log_basename + '.log' try: if os.path.isfile(log_file): filelog = open(log_file,'a') else: filelog = open(log_file,'x') except: print('Unable to Open Log File') sys.exit() BLACK_LIST = [] WHITE_LIST = [] GW_BL = [] IP_BL = [] RunServer(config)