PythonLearn

做一个端口转发程序

注意：仍然是推荐使用现成的程序，比如frp
下面从设计到开发制作出最终成品

设计

单个文件，使用协程，命令行程序，传入监听端口和目标地址。
本地监听端口，收到连接后本地再连接目标地址，之后做转发。
在收到连接、断开连接、连接失败时打印日志。

模块

另外：使用Python3.9版本

开发

安装模块

typer是第三方库，需要安装

pip install typer

1. 准备框架

import typer
import logging
import asyncio


async def handle_conn(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
    ...


async def forward(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
    ...


async def main(
    bind_port: int = typer.Argument(min=1, max=65535),
    target_ip: str = typer.Argument(metavar="IP"),
    target_port: int = typer.Argument(min=1, max=65535),
    bind_ip: str = typer.Option("0.0.0.0", metavar="IP")
):
    """端口转发程序"""


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
    from functools import wraps
    # 方法包装，让typer能够运行协程，这种写法比较神奇，推荐只在个人项目中使用
    typer.run(wraps(main)(lambda *args, **kwargs: asyncio.run(main(*args, **kwargs))))

现在测试一下

PS D:\PythonProjects\Learn\test> python .\temp.py --help            
                                                                    
 Usage: temp.py [OPTIONS] BIND_PORT IP TARGET_PORT                  
                                                                    
 端口转发程序
╭─ Arguments ───────────────────────────────────────────────────────
│ *    bind_port        INTEGER RANGE  [default: None] [required]   
│ *    target_ip        IP             [default: None] [required]   
│ *    target_port      INTEGER RANGE  [default: None] [required]   
╰───────────────────────────────────────────────────────────────────
╭─ Options ─────────────────────────────────────────────────────────
│ --bind-ip        IP  [default: 0.0.0.0]                           
│ --help               Show this message and exit.                  
╰───────────────────────────────────────────────────────────────────

2. 创建监听

注意到handle_conn仅包含两个参数不够，于是改成

async def handle_conn(target_ip: str, target_port: int, reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
    ...

编写main函数

from functools import partial
async def main(
    bind_port: int = typer.Argument(min=1, max=65535),
    target_ip: str = typer.Argument(metavar="IP"),
    target_port: int = typer.Argument(min=1, max=65535),
    bind_ip: str = typer.Option("0.0.0.0", metavar="IP")
):
    """端口转发程序"""
    handler = partial(handle_conn, target_ip, target_port)
    async with await asyncio.start_server(handler, bind_ip, bind_port) as server:
        logging.info(f"开始监听 {bind_ip}:{bind_port}")
        await server.serve_forever()

测试

PS D:\PythonProjects\Learn\test> python .\temp.py 8080 127.0.0.1 8081
2025-08-20 19:48:06,413 - root - INFO - 开始监听 0.0.0.0:8080

Aborted.

3. 转发函数

很简单，读，然后写，如果读到空，就抛错，交给上面处理

async def forward(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
    while True:
        data = await reader.read(4096)
        if not data:
            await writer.drain()
            raise EOFError
        writer.write(data)

4. 处理连接

handle_conn可能遇到两种错误，1是连接目标地址被拒绝，2是连接被关闭，所以需要捕捉并打印

async def handle_conn(target_ip: str, target_port: int, reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
    conn_addr = writer.get_extra_info('peername')
    conn_addr = f"{conn_addr[0]}:{conn_addr[1]}"
    # 连接目标地址
    try:
        target_reader, target_writer = await asyncio.open_connection(target_ip, target_port)
        logging.info(f"{conn_addr}已连接{target_ip}:{target_port}")
    except ConnectionRefusedError as e:  # 连接出错时
        logging.error(f"{conn_addr}连接{target_ip}:{target_port}被拒绝:{e}")
        return
    # 转发
    try:
        await asyncio.gather(
            forward(reader, target_writer),
            forward(target_reader, writer)
        )
    except EOFError:  # 关闭连接时
        logging.info(f"{conn_addr}已断开")
    except Exception as e:  # 转发出错时
        logging.error(f"{conn_addr}与{target_ip}:{target_port}通信异常:{e}")
    finally:
        target_writer.close()
        writer.close()

5. 测试

在本地搭建一个网站，然后试试

PS D:\PythonProjects\Learn\test> python .\temp.py 8081 127.0.0.1 80
2025-08-20 20:29:00,592 - root - INFO - 开始监听 0.0.0.0:8081
2025-08-20 20:29:03,448 - root - INFO - 127.0.0.1:53338已连接127.0.0.1:80
2025-08-20 20:29:03,449 - root - INFO - 127.0.0.1:53339已连接127.0.0.1:80
2025-08-20 20:29:11,806 - root - INFO - 127.0.0.1:53344已连接127.0.0.1:80
2025-08-20 20:29:11,808 - root - INFO - 127.0.0.1:53345已连接127.0.0.1:80
2025-08-20 20:29:11,809 - root - INFO - 127.0.0.1:53347已连接127.0.0.1:80
2025-08-20 20:29:20,211 - root - INFO - 127.0.0.1:53339已断开
2025-08-20 20:29:20,211 - root - INFO - 127.0.0.1:53338已断开
2025-08-20 20:29:20,211 - root - INFO - 127.0.0.1:53344已断开
2025-08-20 20:29:20,211 - root - INFO - 127.0.0.1:53345已断开
2025-08-20 20:29:20,212 - root - INFO - 127.0.0.1:53347已断开

Aborted.

最终项目：端口转发.py