基于asyncio的TCP服务器如何实现远程IP地址过滤？

跨平台获取远程IP的方法 #

不管是CPython还是MicroPython，都可以通过writer.get_extra_info('peername')获取远程连接的地址信息，该方法返回元组(远程IP, 远程端口)，取第一个元素即可得到IP地址。

IP过滤的介入时机 #

在newConnection方法最开始执行——也就是刚建立连接、还未处理任何业务逻辑时，完成IP检查：

1. 获取远程IP地址
2. 用预编译的正则self.ipRange匹配IP
3. 若不匹配则直接关闭连接，终止后续处理

修改后的代码实现 #

async def newConnection(self, reader, writer):
    # 获取远程IP地址
    remote_addr = writer.get_extra_info('peername')
    if remote_addr is None:
        log("无法获取远程地址，拒绝连接")
        writer.close()
        await writer.wait_closed()
        return
    
    remote_ip = remote_addr[0]
    # 检查IP是否符合规则
    if not self.ipRange.match(remote_ip):
        log(f"拒绝来自非法IP的连接: {remote_ip}")
        writer.close()
        await writer.wait_closed()
        return

    # 原有的连接处理逻辑
    log(f"已连接: {remote_ip}")
    self.connection = (reader, writer)
    self.timeout.extend(30)
    while not self.timeout.expired():
        if not await self.pollOpenConnection(reader, writer):
            await asyncio.sleep(1./25.)
            
    writer.close()
    await writer.wait_closed()
    log(f"已断开连接: {remote_ip}")

注意事项 #

• MicroPython连接关闭：部分MicroPython版本中，writer.close()需要配合await writer.wait_closed()确保连接完全释放，避免资源泄漏。
• 正则规则优化：你的正则包含localhost，但peername返回的是IP地址（如127.0.0.1），localhost规则无法匹配实际IP，可移除该条目，保留127.0.0.1即可覆盖本地连接场景。
• 极端情况处理：若peername返回None（极端异常场景），直接拒绝连接，避免后续逻辑报错。

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