package client import ( "bufio" "encoding/binary" "fmt" "io" "log/slog" "net" "proxy-server/client/core" "proxy-server/client/env" "proxy-server/client/geo" "proxy-server/client/report" "proxy-server/pkg/utils" "time" "errors" _ "net/http/pprof" ) func Start() error { // 初始化环境变量 slog.Debug("初始化环境变量...") err := env.Init() if err != nil { return fmt.Errorf("初始化环境变量失败: %w", err) } // 获取归属地 slog.Debug("获取节点归属地...") err = geo.Query() if err != nil { slog.Error("获取归属地失败", "err", err) } // 注册节点 slog.Debug("注册节点...") id, host, err := report.Online(geo.Prov, geo.City, geo.Isp) if err != nil { return fmt.Errorf("注册节点失败: %w", err) } // 性能监控 // go func() { // runtime.SetBlockProfileRate(1) // err := http.ListenAndServe(":7070", nil) // if err != nil { // slog.Error("性能监控服务启动失败", "err", err) // } // }() // 建立控制通道 for { err := ctrl(id, host) if err != nil { slog.Error("建立控制通道失败", "err", err) slog.Info(fmt.Sprintf("%d 秒后重试", core.RetryInterval)) time.Sleep(time.Duration(core.RetryInterval) * time.Second) } } } func ctrl(id int32, host string) error { ctrlAddr := net.JoinHostPort(host, fmt.Sprintf("%d", core.FwdCtrlPort)) dataAddr := net.JoinHostPort(host, fmt.Sprintf("%d", core.FwdDataPort)) slog.Info("建立控制通道", "addr", ctrlAddr) conn, err := net.Dial("tcp", ctrlAddr) if err != nil { return errors.New("连接失败") } defer utils.Close(conn) // 发送客户端信息 var buf = make([]byte, 4) _, err = binary.Encode(buf, binary.BigEndian, id) if err != nil { return fmt.Errorf("编码客户端 ID 失败: %w", err) } _, err = conn.Write(buf) if err != nil { return fmt.Errorf("发送客户端 ID 失败: %w", err) } // 等待服务端响应 reader := bufio.NewReader(conn) respBuf, err := reader.ReadByte() if err != nil { return errors.New("接收响应失败") } if respBuf != 1 { return errors.New("服务端响应失败") } else { slog.Info("成功建立连接") } // 等待用户连接 // 读写失败后退出重连,防止后续数据读写顺序错位导致卡死控制通道 slog.Info("等待用户连接") for { // 接收 dst dstLen, err := reader.ReadByte() if err != nil { return errors.New("接收 dstLen 失败") } dstBuf, err := utils.ReadBuffer(reader, int(dstLen)) if err != nil { return errors.New("接收 dstBuf 失败") } addr := string(dstBuf) // 接收 tag tagLen, err := reader.ReadByte() if err != nil { return errors.New("接收 tagLen 失败") } tagBuf, err := utils.ReadBuffer(reader, int(tagLen)) if err != nil { return errors.New("接收 tagBuf 失败") } // 建立数据通道 go func() { err := data(dataAddr, addr, tagBuf) if err != nil { slog.Error("建立数据通道失败", "err", err) } }() } } func data(dataAddr string, dest string, tag []byte) error { // 向服务端建立连接 src, err := net.Dial("tcp", dataAddr) if err != nil { return errors.New("连接服务端失败") } tagLen := byte(len(tag)) tagBuf := make([]byte, 2+tagLen) tagBuf[1] = tagLen copy(tagBuf[2:], tag) // 向目标地址建立连接 dst, dstErr := net.Dial("tcp", dest) if dstErr != nil { tagBuf[0] = 0 } else { tagBuf[0] = 1 } // 发送连接状态 _, err = src.Write(tagBuf) if err != nil { utils.Close(src) if dst != nil { utils.Close(dst) } return errors.New("发送连接状态失败") } if tagBuf[0] == 0 { utils.Close(src) if dst != nil { utils.Close(dst) } return errors.New("连接目标地址失败") } go func() { defer utils.Close(dst) _, err := io.Copy(dst, src) if err != nil && !errors.Is(err, net.ErrClosed) { slog.Error("上行流量代理失败", "err", err) } }() go func() { defer utils.Close(src) _, err := io.Copy(src, dst) if err != nil && !errors.Is(err, net.ErrClosed) { slog.Error("下行流量代理失败", "err", err) } }() return nil }