0

У меня есть сокет сервер на 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()
9
  • 1
    У вас там что-то приходит, что не может быть раскодировано как UTF-8, судя по ошибке
    – CrazyElf
    Commented 10 мар. 2022 в 8:52
  • 1
    внутри start_my_server сделайте бесконечный цикл и крутите все в нем. А не вызывайте start_my_server() в случае исключения. Так как сейчас у вас при любом исключении получается вложенный (рекурсивный) вызов start_my_server, что расходует память стека, а значит при большом количестве исключений программа израсходует много памяти и в конце концов упадет по переполнению стека
    – Mike
    Commented 10 мар. 2022 в 9:34
  • 1
    И да, вот такой вложенный вызов приводит к тому, что при возникновении исключения во вложенном вызове получается, что оно произошло во время обработки исключения предыдущего уровня, после чего все собственно и падает по приведенной ошибке
    – Mike
    Commented 10 мар. 2022 в 9:36
  • 1
    прямо перед try еще один while True. А в except socket.error просто pass. И еще except UnicodeDecodeError сделать на случай приема кривых данных из за которых сейчас все падает.
    – Mike
    Commented 10 мар. 2022 в 11:36
  • 1
    Но у вас тут еще одна проблема, что если я открою соединение, но никаких данных посылать не буду ? Я так могу оставить открытое соединение хоть на сутки и в это время ваш скрипт будет висеть на recv и управление не возвращать. В итоге других клиентов он обслуживать не сможет.
    – Mike
    Commented 10 мар. 2022 в 11:41

1 ответ 1

0

Сейчас мой код выглядит так, спасибо @Mike за помощь

Осталось решить вопросы с соединением пакетов и #отключением канала по таймауту

    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)
    socket_server.settimeout(4)

    
    # create webserver socket
    def start_my_server():
        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("']", '')
            if secret_token == key:
                socket_server.settimeout(None)                
                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')
            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)
            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()
        while True:
            try:
                while True:
                    print('Working...')
                    client_socket, address = socket_server.accept()
                    print('loop', address)
                    try:
                        data = client_socket.recv(1048576).decode('utf-8')
                        load_secret(secret_token, data)
                    except UnicodeDecodeError:
                        fail_verifivcation()
                    # content = load_page_from_get_request(data)
                    # return data, client_socket, address
                    # load_secret(secret_token, data)
            except socket.error:
                pass
    
    
    def next_step():
        print('next step')
        # start_my_server()
    
    
    print('loop script')
    if __name__ == '__main__':
        print('name = main')
        start_my_server()
        # load_secret()
        next_step()

Ваш ответ

Нажимая «Отправить ответ», вы соглашаетесь с условиями пользования и подтверждаете, что прочитали политику конфиденциальности.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.