Python Socket服务端暴露公网后出现陌生IP连接该如何解决？

• 增加访问鉴权逻辑：你的服务端目前没有任何身份校验机制，任意IP连接后只要发送open/closed字符串就能获取数据库敏感数据，风险极高。建议在通信流程中增加密钥校验环节：提前在Flutter客户端和服务端各存储一个无规律的长字符串密钥，客户端建立连接后优先发送密钥，服务端校验密钥不匹配时直接断开连接，不执行任何后续业务逻辑。
• 限制IP访问范围：如果你的Flutter客户端使用的公网IP固定，可以在路由器端口转发规则或服务器防火墙中配置IP白名单，仅允许指定IP访问2222端口，直接拦截所有陌生IP的连接请求。如果IP不固定，可更换带身份校验能力的内网穿透方案，避免直接将本地端口暴露到公网。
• 补充服务端异常处理逻辑：你当前的代码没有任何异常捕获机制，恶意扫描者发送空数据、乱码数据都会直接导致服务崩溃。建议给网络收发、解码、数据库操作逻辑补充try-except异常捕获，同时增加连接超时机制，连接建立后3秒内未收到有效请求直接断开，避免恶意连接占用服务资源。
• 更换更安全的通信方案：如果没有特殊需求，不建议用裸Socket做对外服务，可以改用轻量Web框架（如Flask、FastAPI）写HTTP接口，自带基础的网络异常处理能力，仅需增加简单的API Key校验即可获得比裸Socket高得多的安全性。
• 更换非默认端口：将当前的2222端口更换为10000以上的冷门端口，可减少大部分公网广谱扫描的流量，降低被恶意攻击的概率。

优化后代码示例 #

from datetime import datetime
import socket
from SqlServerRequest import SqlServerRequest

# 提前配置的校验密钥，和Flutter客户端保持一致
ACCESS_KEY = "自行生成的无规律长字符串"
# 允许访问的IP白名单，不需要可删除
ALLOWED_IPS = {"你的客户端公网IP"}

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(('192.168.43.126', 2222))
s.listen(5)
while True:
    try:
        clientsocket, address = s.accept()
        client_ip = address[0]
        print(f"Connection from {address} has been established.")
        
        # IP白名单校验，不需要可删除
        if client_ip not in ALLOWED_IPS:
            print(f"Reject unknown IP {client_ip}")
            clientsocket.close()
            continue
            
        # 设置3秒超时，防止空连接占用资源
        clientsocket.settimeout(3)
        
        try:
            data = clientsocket.recv(2048).decode().strip()
            print(data)
            
            # 校验逻辑：请求数据格式为 密钥:请求类型 例："xxx123:open"
            if ":" not in data:
                clientsocket.close()
                continue
            req_key, req_type = data.split(":", 1)
            if req_key != ACCESS_KEY:
                print(f"Invalid key from {client_ip}")
                clientsocket.close()
                continue
                
            # 业务逻辑
            if req_type == "open":
                o = SqlServerRequest.dataForOpenOrders()
                clientsocket.sendall(bytes(o,"utf-8"))
            elif req_type == "closed":
                c = SqlServerRequest.dataForClosedOrders()
                clientsocket.sendall(bytes(c,"utf-8"))
        except Exception as e:
            print(f"Handle connection error: {e}")
        finally:
            clientsocket.close()
    except Exception as e:
        print(f"Server error: {e}")

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