From 2b93202789589af5f63799d0e3399ca8adc2eddc Mon Sep 17 00:00:00 2001 From: vikilpet <1+github@4592.ru> Date: Mon, 3 Jan 2022 17:35:57 +0400 Subject: [PATCH] + favicon, asterisk, reload --- readme.md | 20 +++--- readme_ru.md | 10 +-- resources.py | 99 ++++++++++++-------------- rosapi.py | 3 - web_knocking.py | 180 +++++++++++++++++++++++++++--------------------- 5 files changed, 160 insertions(+), 152 deletions(-) diff --git a/readme.md b/readme.md index f257b7f..c706a7b 100644 --- a/readme.md +++ b/readme.md @@ -1,18 +1,18 @@ # Web Knocking for MikroTik -![Screenshot](https://user-images.githubusercontent.com/43970835/79135939-3172f900-7dc1-11ea-9d26-f923c47d47b6.png) +![Screenshot](https://user-images.githubusercontent.com/43970835/147935163-3ebf16d4-4c15-44b2-baec-ee854becf0e4.png) ## How it works -It is a web server written in Python. +This is a web server written in Python -If the request is *good*, then the IP address of the person who made the request is added to the white list on the router. IP addresses of *bad* requests are blacklisted. +If the request is *good* then the IP address of the person who made the request is added to the white list on the router. IP addresses of *bad* requests are blacklisted. -It is assumed that for the white list you allow something, and for the black list you block any incoming connections. +It is assumed that for the white list you allow something and for the black list you block any incoming connections. -For users who need to connect outside the office, we give a link like this: +For users who need to connect outside the office we give a link like this: http://100.100.1.2:2020/access_SeCrEtCoDe -It is possible to support other vendors' devices with help of [netmiko](https://github.com/ktbyers/netmiko), but I only have MikroTik-s. +It is possible to support other vendors' devices with help of [netmiko](https://github.com/ktbyers/netmiko) but I only have MikroTik-s. ## Setup ### Option 1: EXE @@ -21,20 +21,20 @@ Just download the archive from the release page. ### Option 2: Python **Requirements:** Python 3.8; Windows 7+ -It should also work on Linux, but I haven't checked. +It should also work on Linux but I haven't checked. Just download the project. No third-party dependencies. ## Usage Change settings in *settings.ini* to your own. -It is highly recommended to add your IP address to *safe_hosts* to avoid blacklisting during testing. +It is highly recommended to add your IP address to *safe_hosts* to avoid blacklisting during testing. From the addresses on the *safe_hosts* list you can request a reload of the settings with url `http://ip:port/reload` Forward a port in the router from the settings to the computer where Web Knocking is running. Enable API access on the router from this computer (*ip services - api-ssl* or *api*). -For white and black lists, make the rules according to your needs. For example, forward port 80 for white list on MikroTik: +For white and black lists make the rules according to your needs. For example forward port 80 for white list on MikroTik: /ip firewall nat add src-address-list=white_list in-interface=WAN \ dst-port=80 action=dst-nat to-addresses=192.168.0.10 to-ports=80 @@ -42,7 +42,7 @@ For white and black lists, make the rules according to your needs. For example, ### Only in Windows: In console window: - press «i» to print out IP addresses; -- «u» to print out users; +- «u» to print out users (expired dates are marked with an asterisk); - «s» to reload settings so you can add new users without restart. ## Support project diff --git a/readme_ru.md b/readme_ru.md index a3e3782..882a91b 100644 --- a/readme_ru.md +++ b/readme_ru.md @@ -1,10 +1,10 @@ # Web Knocking для MikroTik -![Screenshot](https://user-images.githubusercontent.com/43970835/79136043-63845b00-7dc1-11ea-9e30-6d4e011c2bc3.png) +![Screenshot](https://user-images.githubusercontent.com/43970835/147935163-3ebf16d4-4c15-44b2-baec-ee854becf0e4.png) ## Как это работает Это веб-сервер на Python. -Если запрос *правильный*, то IP-адрес того, кто сделал этот запрос, добавляется в белый список на маршрутизаторе. IP-адреса *плохих* запросов попадают в чёрный список. +Если запрос *правильный*, то IP-адрес того, кто сделал этот запрос, добавляется в *белый список* на маршрутизаторе. IP-адреса *плохих* запросов попадают в *чёрный список*. Предполагается, что для белого списка вы что-то разрешаете, а для чёрного списка блокируются любые входящие соединения. @@ -26,9 +26,9 @@ Просто скачайте [проект](https://github.com/vikilpet/Web-Knocking/releases). Без зависимостей от нестандартных модулей. ## Использование -Измените настройки в settings.ini на свои. +Измените настройки в *settings.ini* на свои. -Очень рекомендуется добавить свой IP адрес в *safe_hosts*, чтобы при тестировании не занести себя в чёрный список. +Очень рекомендуется добавить свой IP адрес в *safe_hosts*, чтобы при тестировании не занести себя в чёрный список. С адресов из списка *safe_hosts* можно выполнить запрос на перезагрузку настроек вида `http://ip:port/reload` Пробросьте в маршрутизаторе порт из настроек на компьютер, на котором запущен Web Knocking. @@ -42,7 +42,7 @@ ### Только для Windows: В консоли: - нажмите «i» для вывода списка IP адресов; -- «u» для вывода списка пользователей +- «u» для вывода списка пользователей (истекшие даты помечены звездочкой); - «s» для перезагрузки настроек, чтобы добавлять пользователей без перезапуска. ## Помощь проекту diff --git a/resources.py b/resources.py index 5d4e19a..b50f5a1 100644 --- a/resources.py +++ b/resources.py @@ -7,18 +7,20 @@ def __init__(self, language:str='en'): di = dict(v.split('=') for v in lang_str[:-1].split('\n')) self.__dict__.update(di) +FAVICON = b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x10\x00\x00\x00\x10\x08\x03\x00\x00\x00(-\x0fS\x00\x00\x00\x81PLTE\xff\xff\xff\x83\x83\x82\xa7\xa7\xa6~\x7f}qrp\xfd\xfd\xfc\xe1\xe1\xe0\xde\xde\xde\xba\xba\xb9\x95\x95\x94\x88\x89\x87\xf1\xf1\xf1\xeb\xeb\xea\xe6\xe6\xe5\xd1\xd1\xd1\xce\xce\xcd\x92\x92\x91xxw\xf9\xf9\xf9\xf5\xf5\xf4\xee\xee\xed\xe3\xe3\xe3\xca\xca\xc9\xc2\xc2\xc1\xbc\xbc\xbb\xb2\xb2\xb1\xad\xad\xac\x9f\x9f\x9e\x9a\x9a\x99\x97\x97\x97\x8e\x8e\x8c||{zzyijh\xa4\xa4\xa3\xa3\xa3\xa3\xa1\xa1\xa1\x80\x80\x80ddb^^\\UUTRRPAA@n\xda.\xc4\x00\x00\x00\x92IDAT\x18\xd3}\x8fG\x0e\xc30\x0c\x04IIVq\x97{/\xe9\xc9\xff\x1f\x18\xc92\x82\xf8\xe2\x05\x88\xe5\xcc\x81\x00\xe1,\xaa?r\x92\xd6\xc3A\xa4\xf1]\x15n\x8d\x14I|l\xe9\xd8H\x8b\x01NE\xcfb\xc06\xe1\xdcp\xaeC \xd9+\xe0\x0c\\F\x01\xd1\xa7\xac\x9e\xe9\xd08\xb1\x00\x84\xbe^K\x90\xc4\tj\xc6\x17\xa8\xe6\xeb\xb2\xb1\x87V\xc0\xa5\xf6\xe4~d6S\x81x\xd0nu\x02\x83\xad\x98d\x93\x13!\xcdmu\xef\x18\xf6D\x14\xc9M\x13\xef\xef\x8bL\xf0\x1f~\x01c\xeb\x06\x95_\x99Rn\x00\x00\x00\x00IEND\xaeB`\x82' + dictionary_en='''\ homepage=Homepage: https://github.com/vikilpet/ -donate=Donate, if you like it: https://www.paypal.me/vikil +donate=Donate if you like it: https://www.paypal.me/vikil time_capt=Time ip_capt=Your IP page_title=Knock-knock -access_error={}
there is some error -pass_expired={}
your passcode expired: {} +access_error=You are logged in as «{}»
There is some error :( +pass_expired=You are logged in as «{}»
Your passcode has expired: {} pass_unknown=Unknown passcode ban=Ban -perm_timeout_text={}
access granted -temp_timeout_text={}
access granted for 8 hours +perm_timeout_text=You are logged in as «{}»
Access granted +temp_timeout_text=You are logged in as «{}»
Access granted for 8 hours ''' dictionary_ru='''\ @@ -27,64 +29,53 @@ def __init__(self, language:str='en'): time_capt=Время ip_capt=Ваш IP page_title=Тук-тук -access_error={}
какая-то ошибка -pass_expired={}
ваш код доступа уже истёк: {} +access_error=Вы вошли как «{}»
Какая-то ошибка включения доступа :( +pass_expired=Вы вошли как «{}»
Ваш код доступа уже истёк: {} pass_unknown=Неизвестный код доступа ban=Бан -perm_timeout_text={}
доступ открыт -temp_timeout_text={}
доступ открыт на 8 часов +perm_timeout_text=Вы вошли как «{}»
Доступ открыт +temp_timeout_text=Вы вошли как «{}»
Доступ открыт на 8 часов ''' HTML_DEFAULT = '''\ - + {page_title} - - - + + -
-
- {message}
-
-

- {ip_capt}: {ip_address}
- {time_capt}: {timestamp}
- © https://github.com/vikilpet/
-

-
+
+ {message} +
+

+ {ip_capt}: {ip_address} +
{time_capt}: {timestamp} +
© https://github.com/vikilpet/ +
+

- ''' \ No newline at end of file diff --git a/rosapi.py b/rosapi.py index 8708b37..1654da7 100644 --- a/rosapi.py +++ b/rosapi.py @@ -1,4 +1,3 @@ -# source: https://wiki.mikrotik.com/wiki/Manual:API_Python3 import sys, binascii, socket, select, ssl import hashlib @@ -174,7 +173,6 @@ def main(): passw = "" secure = True port = 0 - if len(sys.argv) == 4: user = sys.argv[2] passw = sys.argv[3] @@ -203,7 +201,6 @@ def main(): if sys.stdin in r[0]: l = sys.stdin.readline() l = l[:-1] - if l == '': apiros.writeSentence(inputsentence) inputsentence = [] diff --git a/web_knocking.py b/web_knocking.py index fe42c21..e5420b6 100644 --- a/web_knocking.py +++ b/web_knocking.py @@ -18,14 +18,12 @@ SetConsoleTitleW = ctypes.windll.kernel32.SetConsoleTitleW TITLE = 'Web Knocking' -APP_VERSION = 'v2020-12-06' +APP_VERSION = 'v2022-01-03' INI_FILE = 'web_knocking.ini' DEF_DEVICE_TYPE = 'mikrotik_routeros' PASS_SEP = '_' event_counter = 0 - - sett = None { '1.2.3.4' : { @@ -103,7 +101,7 @@ class Settings: Convert True/yes to bool. Convert digits to int ''' - def __init__(s, keep_setting_case:bool=False): + def __init__(self, keep_setting_case:bool=False): config = configparser.ConfigParser() config.optionxform = str try: @@ -113,13 +111,12 @@ def __init__(s, keep_setting_case:bool=False): except FileNotFoundError: print(f'{INI_FILE} file not found' + '\nPress any key to exit') - sections = { s.lower() : config._sections[s] for s in config._sections } for section in sections: - setattr(s, section, {}) + setattr(self, section, {}) di = sections[section] if keep_setting_case: settings_names = di.keys() @@ -127,16 +124,16 @@ def __init__(s, keep_setting_case:bool=False): settings_names = [ k.lower() for k in di] for key in settings_names: - if di[key].lower() in ['true', 'yes']: - getattr(s, section)[key] = True - elif di[key].lower() in ['false', 'no']: - getattr(s, section)[key] = False + if di[key].lower() in ('true', 'yes'): + getattr(self, section)[key] = True + elif di[key].lower() in ('false', 'no'): + getattr(self, section)[key] = False elif di[key].isdigit(): - getattr(s, section)[key] = int(di[key]) + getattr(self, section)[key] = int(di[key]) else: - getattr(s, section)[key] = di[key] + getattr(self, section)[key] = di[key] - def __getattr__(s, name): + def __getattr__(self, name): try: log.debug(f'Settings: unknown key: {name}') except: @@ -212,7 +209,6 @@ def process_ip(ip:str, behavior:str reason - if none then ban immediately without checking 'black_threshold' ''' - if not reason: reason = behavior if ip in sett.ips and user: sett.ips[ip]['user'] = user if behavior != 'good' \ @@ -299,6 +295,7 @@ def decision(path:str, ip:str)->list: on success or (False, 'error text') on exception. ''' + global sett try: message = '' if path == '/': @@ -411,21 +408,39 @@ def decision(path:str, ip:str)->list: else: process_ip(ip, behavior='danger', reason='status unsafe') message = lang.ban + elif path == '/reload': + if ip in sett.general['safe_hosts']: + status, data = load_settings() + if status: + log.info(ip.ljust(15), 'settings reloaded') + sett = data + print_users() + message = 'reloaded' + else: + message = 'error' + log.error('failed to reload settings:', data) + else: + process_ip(ip, behavior='danger', reason='reload unsafe') + message = lang.ban else: process_ip(ip, behavior='danger', reason='wrong path') message = lang.ban return True, message except Exception as e: process_ip(ip, behavior='bad', reason='decision error') + log.debug(f'line: {e.__traceback__.tb_lineno}') return False, repr(e) def print_users(): 'Print the dict of users as a table' - table = [ ['User', 'Last day', 'Last IP', 'Last Access'] ] + table = [ ['User', 'Last Day', 'Last IP', 'Last Access'] ] for user in sett.users.values(): + if last_day := user.get('last_day'): + if dt.strptime(last_day, '%Y-%m-%d') < dt.now(): + last_day = '*' + last_day table.append([ user['name'] - , user.get('last_day', None) + , last_day , user['ips'][-1] if user['ips'] else None , user['last_access'].strftime('%Y.%m.%d %H:%M:%S') \ if user['last_access'] else None @@ -444,122 +459,121 @@ def print_ips(): table_print(table, use_headers=True, sorting=[0, 1]) class KnockHandler(BaseHTTPRequestHandler): - def handle_one_request(s): + def handle_one_request(self): try: super().handle_one_request() - req_type = str(s.raw_requestline + req_type = str(self.raw_requestline , encoding='iso-8859-1').split()[0].upper() if req_type != 'GET': - process_ip(s.address_string() + process_ip(self.address_string() , behavior='danger', reason='wrong request method') except ConnectionResetError: - log.debug(s.address_string().ljust(15) + log.debug(self.address_string().ljust(15) , 'connection reset') - process_ip(s.address_string(), behavior='bad' + process_ip(self.address_string(), behavior='bad' , reason='port scan') except IndexError: try: - rr = str(s.raw_requestline, encoding='iso-8859-1') + rr = str(self.raw_requestline, encoding='iso-8859-1') except Exception as e: log.debug( - s.address_string().ljust(15) + self.address_string().ljust(15) , 'raw_requestline error: ' + repr(e)) else: log.debug( - s.address_string().ljust(15) + self.address_string().ljust(15) , 'raw_requestline: {} (len={})'.format( rr, len(rr) ) ) except Exception as e: log.debug( - s.address_string().ljust(15) - , 'h_o_r exception: ' + (repr(e)[:40]) + self.address_string().ljust(15) + , 'h_o_r exception: ' + (repr(e)) \ + + f'\n\tat line: {e.__traceback__.tb_lineno}' ) - process_ip(s.address_string() + process_ip(self.address_string() , behavior='danger', reason='h_o_r exception') - def send_error(s, code, message=None + def send_error(self, code, message=None , explain=None): if code > 500: - s.send_response(200) - s.send_header('Content-type','text/html; charset=utf-8') - s.end_headers() - s.wfile.write(bytes(lang.ban, 'utf-8')) + self.send_response(200) + self.send_header('Content-type','text/html; charset=utf-8') + self.end_headers() + self.wfile.write(bytes(lang.ban, 'utf-8')) else: super().send_error(code, message, explain) - def do_GET(s): - if 'favicon.' in s.path: - log.debug(s.address_string().ljust(15) - , f'favicon request: {s.path}') - s.wfile.write(b'') + def do_GET(self): + if 'favicon.' in self.path: + self.wfile.write(sett.favicon) return - s.send_response(200) - s.send_header('Content-type','text/html; charset=utf-8') - s.end_headers() - status, data = decision(s.path, s.address_string() ) + self.send_response(200) + self.send_header('Content-type','text/html; charset=utf-8') + self.end_headers() + status, data = decision(self.path, self.address_string() ) if status: message = data else: log.debug( - s.address_string().ljust(15) + self.address_string().ljust(15) , 'decision error: ' + data ) message = lang.ban - if s.path == '/status': + if self.path == '/status': page = message else: page = sett.html.format( message=message , timestamp = dt.now() \ .strftime('%Y.%m.%d %H:%M:%S') - , ip_address = s.address_string() + , ip_address = self.address_string() , ip_capt = lang.ip_capt , time_capt = lang.time_capt , page_title = lang.page_title ).replace('\n', '').replace('\t', '') - s.wfile.write(bytes(page, 'utf-8')) + self.wfile.write(bytes(page, 'utf-8')) - def log_message(s, msg_format, *args): + def log_message(self, msg_format, *args): global event_counter if args[0].startswith('GET /status') \ + or args[0].startswith('GET /reload') \ and not sett.general['developer']: return event_counter += 1 set_title(event_counter) log.http( - s.address_string().ljust(15) + self.address_string().ljust(15) , ' '.join(args) ) def load_settings()->tuple: - ''' Load settings from .ini file. - ''' + ''' Load settings from .ini file. ''' + global sett try: - sett = Settings(keep_setting_case=True) + new_sett = Settings(keep_setting_case=True) for opt in DEF_OPT_GENERAL: - sett.general.setdefault(*opt) + new_sett.general.setdefault(*opt) for opt in DEF_OPT_DEVICE: - sett.device.setdefault(*opt) - sett.ips = {} - if isinstance(sett.general['safe_hosts'], str): - ip_string = sett.general['safe_hosts'] - sett.general['safe_hosts'] = [] + new_sett.device.setdefault(*opt) + new_sett.ips = {} + if isinstance(new_sett.general['safe_hosts'], str): + ip_string = new_sett.general['safe_hosts'] + new_sett.general['safe_hosts'] = [] for ip in ip_string.split(','): - sett.general['safe_hosts'].append( + new_sett.general['safe_hosts'].append( ip.strip() ) - users_di = sett.users - sett.users = {} + users_di = new_sett.users + new_sett.users = {} for user in users_di: if ' ' in users_di[user]: - passcode, last_day = \ - users_di[user].split() + passcode, last_day = users_di[user].split() else: passcode = users_di[user] last_day = None - sett.users[user] = { + new_sett.users[user] = { 'name' : user , 'passcode' : passcode , 'last_access' : None @@ -569,19 +583,31 @@ def load_settings()->tuple: if os.path.exists('files/index.html'): with open('files/index.html' , encoding='utf-8') as fd: - sett.html = fd.read() + new_sett.html = fd.read() else: - sett.html = resources.HTML_DEFAULT - if sett.device['device_type'] == \ - DEF_DEVICE_TYPE: - sett.rosapi_args = { - 'ip' : sett.device['host'] - , 'port' : sett.device['port'] - , 'username' : sett.device['username'] - , 'password' : sett.device['password'] - , 'secure' : sett.device['secure'] + new_sett.html = resources.HTML_DEFAULT + if os.path.exists('files/favicon.png'): + with open('files/favicon.png' + , encoding='rb') as fd: + new_sett.favicon = fd.read() + else: + new_sett.favicon = resources.FAVICON + if new_sett.device['device_type'] == DEF_DEVICE_TYPE: + new_sett.rosapi_args = { + 'ip' : new_sett.device['host'] + , 'port' : new_sett.device['port'] + , 'username' : new_sett.device['username'] + , 'password' : new_sett.device['password'] + , 'secure' : new_sett.device['secure'] } - return True, sett + if sett: + new_sett.ips = sett.ips + for u in sett.users: + if new_sett.users.get(u): + new_sett.users[u]['last_access'] = \ + sett.users[u]['last_access'] + new_sett.users[u]['ips'] = sett.users[u]['ips'] + return True, new_sett except Exception as e: return False, repr(e) @@ -688,19 +714,13 @@ def key_wait(): log.info('reload settings') status, data = load_settings() if status: - data.ips = sett.ips - for u in sett.users: - if data.users.get(u): - data.users[u]['last_access'] = \ - sett.users[u]['last_access'] - data.users[u]['ips'] = sett.users[u]['ips'] sett = data print_users() else: log.error('failed to reload settings:', data) time.sleep(0.001) - threading.Thread(target=key_wait, daemon=True).start() + status, sett = load_settings() if not status: print('Error loading settings:', sett)