如何在Python Socket Server的函数外部实现关闭操作？

要在console_win函数外部完成Socket服务器的关闭，核心是让服务器的状态（socket实例、运行状态）能被外部指令访问到，同时解决socket阻塞导致无法响应关闭的问题。以下是具体修改方案：

关键修改点 #

• 移置服务器状态到外层作用域：将socket实例、运行标志放在backdoor_winfunc函数内，避免全局变量污染，让console_win内部函数可直接访问。
• 添加超时与停止标志：给服务器socket设置超时，配合threading.Event作为停止标志，解决s.accept()的阻塞问题，让服务器能响应关闭指令。
• 实现"end server"指令逻辑：在指令处理分支中触发停止标志，关闭socket并清理资源。

修改后的完整代码 #

from tkinter import *
import socket
from tkinter import messagebox
import threading
from PIL import ImageTk, Image

# configurations
bg_color = "#2A2A2A"
frame_bg_color = "#1F1F1F"
label_fg_color = "cyan"
btn_fg_color = "cyan"
btn_bg_color = "#1F1F1F"
entry_bg_color = "#1F1F1F"
entry_fg_color = "red"

def backdoor_winfunc():
    # 服务器状态变量，放在外层作用域，让console_win内部可访问
    server_socket = None
    server_running = threading.Event()
    server_running.clear()  # 初始为未运行状态

    def on_enter(e):
        e.widget["foreground"] = "#fb1466"

    def on_leave(e):
        e.widget["foreground"] = "cyan"

    def connect_internet():
        try:
            sock = socket.create_connection(("www.google.com", 80))
            if sock is not None:
                sock.close()  # 修复原代码语法错误，close需加括号
                return True
        except OSError:
            return False

    def console_win():
        def create_server():
            nonlocal server_socket, server_running
            ip = bck_ip_var.get()
            port = bck_port_var.get()

            if "." not in ip:
                messagebox.showerror("Error", "Invalid ip address!")
                return False

            server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            server_socket.settimeout(1)  # 设置1秒超时，让循环能检查停止标志
            server_socket.bind((ip, port))
            console_box.insert(END, f"[+] Server Started at {ip}:{port}!\n")
            server_socket.listen(1)
            server_running.set()  # 标记服务器运行中

            while server_running.is_set():
                try:
                    conn, addr = server_socket.accept()
                    console_box.insert(END, f"Connected to {addr}!\n")
                    conn.settimeout(1)
                    while server_running.is_set():
                        try:
                            data = conn.recv(1024).decode()
                            if not data or data == "end":
                                conn.close()
                                console_box.insert(END, f"Disconnected from {addr}!\n")
                                break
                        except socket.timeout:
                            continue
                except socket.timeout:
                    continue
            # 关闭服务器socket
            server_socket.close()
            console_box.insert(END, "[-] Server Closed!\n")
            server_socket = None

        def take_cmd(e):
            nonlocal server_socket, server_running
            cmd = cmd_box_var.get()

            if cmd == "help":
                console_box.config(fg="red")
                console_box.config(fg="lime")
                try:
                    with open("Assets\\cmd_files\\help.txt", "r") as rw:
                        help_data = rw.read()
                        console_box.insert(END, help_data + "\n\n")
                except FileNotFoundError:
                    console_box.insert(END, "[-] Help file not found!\n")

            elif cmd == "clear":
                console_box.delete("1.0", END)

            elif cmd == "start server":
                if not server_running.is_set():
                    t1 = threading.Thread(target=create_server, daemon=True)  # 设置守护线程，避免程序残留
                    t1.start()
                else:
                    console_box.insert(END, "[-] Server is already running!\n")

            elif cmd == "end server":
                if server_running.is_set() and server_socket:
                    server_running.clear()
                    console_box.insert(END, "[*] Stopping server...\n")
                else:
                    console_box.insert(END, "[-] Server is not running!\n")

            else:
                console_box.insert(
                    END,
                    "\n[-] Invalid command! please check our help menu using command 'help'\n",
                )
            # 清空输入框
            cmd_box_var.set("")

        # check internet connection
        is_internet = connect_internet()
        if is_internet == False:
            messagebox.showerror(
                "Error", "Oops! Its look like you are not connected to internet"
            )
            return False

        # configurations
        console_win = Toplevel()
        console_win.geometry("800x605+200+50")
        console_win.title("Console")
        console_win.config(bg="#252525")

        scrl_bar = Scrollbar(console_win, width=20)
        scrl_bar.pack(side=RIGHT, fill=Y)

        console_box = Text(
            console_win,
            height=19,
            width=60,
            bg=frame_bg_color,
            state="normal",
            yscrollcommand=scrl_bar.set,
            font=("Arial"),
            fg="lime",
        )
        console_box.pack()

        scrl_bar.config(command=console_box.yview)

        cmd_box_var = StringVar(value="start server")

        command_box = Entry(
            console_win,
            font=("Arial 30"),
            textvariable=cmd_box_var,
            bg=bg_color,
            fg="cyan",
            width=30,
        )
        command_box.focus()
        command_box.bind("<Return>", take_cmd)
        command_box.pack(pady=(10, 0))

        console_win.mainloop()

    backdoorwin = Tk()
    backdoorwin.config(bg=bg_color)
    backdoorwin.title("Backdoor")
    backdoorwin.geometry("800x600+200+50")
    # frame
    backdoor_f = Frame(backdoorwin, bg=frame_bg_color)
    backdoor_f.pack(pady=120)
    # header
    img = Image.open("Assets\\images\\death.png")
    img = img.resize((100, 100))
    logo = ImageTk.PhotoImage(img)
    logo_lb = Label(backdoor_f, image=logo, bg=frame_bg_color)
    logo_lb.grid(row=0, column=0, pady=(15, 15))
    bck_logo_txt = Label(
        backdoor_f,
        text="BACKDOOR",
        fg="#fb1466",
        bg=frame_bg_color,
        font=("sarif 40 bold"),
    )
    bck_logo_txt.grid(row=0, column=1, pady=(15, 0), padx=(0, 0))

    # variables
    bck_ip_var = StringVar(value=socket.gethostbyname(socket.gethostname()))
    bck_port_var = IntVar(value=8080)

    # label
    bck_ip_lb = Label(
        backdoor_f,
        text="Ip Address",
        font=("sarif 20 bold"),
        bg=frame_bg_color,
        fg=label_fg_color,
    )
    bck_ip_lb.grid(row=1, column=0, padx=(60, 30), pady=(0, 0))
    bck_port_lb = Label(
        backdoor_f,
        text="Port",
        font=("sarif 20 bold"),
        bg=frame_bg_color,
        fg=label_fg_color,
    )
    bck_port_lb.grid(row=2, column=0, padx=(60, 30), pady=(15, 15))
    # entry
    bck_ip_en = Entry(
        backdoor_f,
        textvariable=bck_ip_var,
        font=("san-sarif 20"),
        bg=entry_bg_color,
        fg=entry_fg_color,
        justify="center",
    )
    bck_ip_en.grid(row=1, column=1, padx=(0, 30), pady=(0, 0))

    bck_port_en = Entry(
        backdoor_f,
        textvariable=bck_port_var,
        font=("san-sarif 20"),
        bg=entry_bg_color,
        fg=entry_fg_color,
        justify="center",
    )
    bck_port_en.grid(row=2, column=1, padx=(0, 30), pady=(15, 15))

    # button
    cr_backdoor_btn = Button(
        backdoor_f,
        text="Host",
        font=("sarif 20 bold"),
        bg=btn_bg_color,
        fg=btn_fg_color,
        cursor="hand2",
        command=lambda: console_win(),
    )
    cr_backdoor_btn.grid(row=3, column=1, pady=(0, 70))
    # btnconfig
    cr_backdoor_btn.bind("<Enter>", on_enter)
    cr_backdoor_btn.bind("<Leave>", on_leave)
    backdoorwin.mainloop()


if __name__ == "__main__":
    backdoor_winfunc()

额外说明 #

• 修复了原代码中sock.close缺少括号的语法错误。
• 给服务器线程设置了daemon=True，确保主程序退出时线程能自动结束。
• 在"end server"指令后清空输入框，提升交互体验。
• 处理了帮助文件不存在的异常情况，避免程序崩溃。

内容的提问来源于stack exchange，提问作者DITROX