У меня есть сокет сервер на pyhon3 с некой секьюрностью из-за гуляющих ботов которые стучатся во все открытые порты Скрипт работает исправно, только вот из-за Dos падает с ошибкой
Код ошибки:
During handling of the above exception, another exception occurred:
Traceback (most recent call last): File "SM_803GIT.py", line 94, in <module>
start_my_server() File "SM_803GIT.py", line 79, in start_my_server
load_secret(secret_token, data) File "SM_803GIT.py", line 43, in load_secret
accept(key) File "SM_803GIT.py", line 49, in accept
try_to_verification(key) File "SM_803GIT.py", line 60, in try_to_verification
next_step() File "SM_803GIT.py", line 89, in next_step
start_my_server() File "SM_803GIT.py", line 79, in start_my_server
load_secret(secret_token, data) File "SM_803GIT.py", line 43, in load_secret
accept(key) File "SM_803GIT.py", line 49, in accept
try_to_verification(key) File "SM_803GIT.py", line 60, in try_to_verification
next_step() File "SM_803GIT.py", line 89, in next_step
start_my_server() File "SM_803GIT.py", line 79, in start_my_server
load_secret(secret_token, data) File "SM_803GIT.py", line 43, in load_secret
accept(key) File "SM_803GIT.py", line 49, in accept
try_to_verification(key) File "SM_803GIT.py", line 60, in try_to_verification
next_step() File "SM_803GIT.py", line 89, in next_step
start_my_server() File "SM_803GIT.py", line 79, in start_my_server
load_secret(secret_token, data) File "SM_803GIT.py", line 43, in load_secret
accept(key) File "SM_803GIT.py", line 49, in accept
try_to_verification(key) File "SM_803GIT.py", line 60, in try_to_verification
next_step() File "SM_803GIT.py", line 89, in next_step
start_my_server() File "SM_803GIT.py", line 79, in start_my_server
load_secret(secret_token, data) File "SM_803GIT.py", line 43, in load_secret
accept(key) File "SM_803GIT.py", line 49, in accept
try_to_verification(key) File "SM_803GIT.py", line 60, in try_to_verification
next_step() File "SM_803GIT.py", line 89, in next_step
start_my_server() File "SM_803GIT.py", line 79, in start_my_server
load_secret(secret_token, data) File "SM_803GIT.py", line 43, in load_secret
accept(key) File "SM_803GIT.py", line 49, in accept
try_to_verification(key) File "SM_803GIT.py", line 60, in try_to_verification
next_step() File "SM_803GIT.py", line 89, in next_step
start_my_server() File "SM_803GIT.py", line 79, in start_my_server
load_secret(secret_token, data) File "SM_803GIT.py", line 43, in load_secret
accept(key) File "SM_803GIT.py", line 49, in accept
try_to_verification(key) File "SM_803GIT.py", line 60, in try_to_verification
next_step() File "SM_803GIT.py", line 89, in next_step
start_my_server() File "SM_803GIT.py", line 79, in start_my_server
load_secret(secret_token, data) File "SM_803GIT.py", line 43, in load_secret
accept(key) File "SM_803GIT.py", line 49, in accept
try_to_verification(key) File "SM_803GIT.py", line 60, in try_to_verification
next_step() File "SM_803GIT.py", line 89, in next_step
start_my_server() File "SM_803GIT.py", line 81, in start_my_server
start_my_server() File "SM_803GIT.py", line 76, in start_my_server
data = client_socket.recv(1048576).decode('utf-8') UnicodeDecodeError: 'utf-8' codec can't decode byte 0xee in position 4: invalid continuation byte
Как мне обработать такие случаи или попросту пропускать большое кол-во запросов от одного ip Пример кода:
import os
import socket
from pathlib import Path
from dotenv import load_dotenv
import json
import time
import re
#Init .env
env_path = Path('.') / '.env'
load_dotenv(dotenv_path=env_path)
#init
secret_token = os.environ['secret_token']
#time
seconds = time.time()
local_time = time.ctime(seconds)
#HDRS
HDRS = 'HTTP/1.1 200 OK\r\nContent-Type: text/html; charset=utf-8\r\n\r\n'
HDRS_404 = 'HTTP/1.1 404 OK\r\nContent-Type: text/html; charset=utf-8\r\n\r\n'
socket_server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket_server.bind(('ip', port))
socket_server.listen(32768)
#create webserver socket
def start_my_server():
# print('Working...')
# client_socket, address = socket_server.accept()
# print('loop', address)
#
# data = client_socket.recv(1048576).decode('utf-8')
def load_secret(secret_token, data):
key = str(data)
key = re.findall(f'X-Gitlab-Token:.............', key)
key = str(key).replace("['X-Gitlab-Token: ",'')
key = str(key).replace("']", '')
# print(key, 'load_secret')
# print(secret_token, 'load_secret')
if secret_token == key:
accept(key)
else:
fail_verifivcation()
# return key
def accept(key):
print('Find')
try_to_verification(key)
def try_to_verification(key):
print(key, 'key try_to_verification')
# print(data,'\n', client_socket,'\n', address,'\n', 'ПРОВЕРКА')
with open(f'data("{local_time}").json', 'w') as output_file:
json.dump(data, output_file)
client_socket.send(HDRS.encode('utf-8'))
client_socket.shutdown(socket.SHUT_WR)
# socket_server.close()
# return init_next()
next_step()
def fail_verifivcation():
print('Not find')
client_socket.shutdown(socket.SHUT_WR)
addresses = open('ipPOST', 'a')
addresses.write(str(address) + f'{local_time}\n')
addresses.close()
# init_next()
try:
while True:
print('Working...')
client_socket, address = socket_server.accept()
print('loop', address)
addresses = open('ipPOST', 'a')
addresses.write(str(address) + f'{local_time}\n')
# addresses.close()
data = client_socket.recv(1048576).decode('utf-8')
# content = load_page_from_get_request(data)
# return data, client_socket, address
load_secret(secret_token, data)
except socket.error:
start_my_server()
# load_secret(secret_token, data)
# init_next()
# return init_next()
def next_step():
print('next step')
# return init_next()
start_my_server()
print('loop script')
if __name__ == '__main__':
print('name = main')
start_my_server()
# load_secret()
next_step()