主页 / 问题 / 1388880
Accepted
Miekrif
Asked:2022-08-10 15:37:03 +0000 UTC2022-08-10 15:37:03 +0000 UTC

使用 Python 套接字进行异常处理或故障排除

  • 772

我在 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()
python

1 个回答

  1. Best Answer

    现在我的代码看起来像这样，感谢@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()
    • 0

相关问题