package services import ( "bufio" "fmt" "hash/crc32" "io" "log" "net" "github.com/snail007/goproxy/utils" "runtime/debug" "strconv" "strings" "time" ) type UDP struct { p utils.ConcurrentMap outPool utils.OutPool cfg UDPArgs sc *utils.ServerChannel } func NewUDP() Service { return &UDP{ outPool: utils.OutPool{}, p: utils.NewConcurrentMap(), } } func (s *UDP) InitService() { if *s.cfg.ParentType != TYPE_UDP { s.InitOutConnPool() } } func (s *UDP) StopService() { if s.outPool.Pool != nil { s.outPool.Pool.ReleaseAll() } } func (s *UDP) Start(args interface{}) (err error) { s.cfg = args.(UDPArgs) if *s.cfg.Parent != "" { log.Printf("use %s parent %s", *s.cfg.ParentType, *s.cfg.Parent) } else { log.Fatalf("parent required for udp %s", *s.cfg.Local) } s.InitService() host, port, _ := net.SplitHostPort(*s.cfg.Local) p, _ := strconv.Atoi(port) sc := utils.NewServerChannel(host, p) s.sc = &sc err = sc.ListenUDP(s.callback) if err != nil { return } log.Printf("udp proxy on %s", (*sc.UDPListener).LocalAddr()) return } func (s *UDP) Clean() { s.StopService() } func (s *UDP) callback(packet []byte, localAddr, srcAddr *net.UDPAddr) { defer func() { if err := recover(); err != nil { log.Printf("udp conn handler crashed with err : %s \nstack: %s", err, string(debug.Stack())) } }() var err error switch *s.cfg.ParentType { case TYPE_TCP: fallthrough case TYPE_TLS: err = s.OutToTCP(packet, localAddr, srcAddr) case TYPE_UDP: err = s.OutToUDP(packet, localAddr, srcAddr) default: err = fmt.Errorf("unkown parent type %s", *s.cfg.ParentType) } if err != nil { log.Printf("connect to %s parent %s fail, ERR:%s", *s.cfg.ParentType, *s.cfg.Parent, err) } } func (s *UDP) GetConn(connKey string) (conn net.Conn, isNew bool, err error) { isNew = !s.p.Has(connKey) var _conn interface{} if isNew { _conn, err = s.outPool.Pool.Get() if err != nil { return nil, false, err } s.p.Set(connKey, _conn) } else { _conn, _ = s.p.Get(connKey) } conn = _conn.(net.Conn) return } func (s *UDP) OutToTCP(packet []byte, localAddr, srcAddr *net.UDPAddr) (err error) { numLocal := crc32.ChecksumIEEE([]byte(localAddr.String())) numSrc := crc32.ChecksumIEEE([]byte(srcAddr.String())) mod := uint32(*s.cfg.PoolSize) if mod == 0 { mod = 10 } connKey := uint64((numLocal/10)*10 + numSrc%mod) conn, isNew, err := s.GetConn(fmt.Sprintf("%d", connKey)) if err != nil { log.Printf("upd get conn to %s parent %s fail, ERR:%s", *s.cfg.ParentType, *s.cfg.Parent, err) return } if isNew { go func() { defer func() { if err := recover(); err != nil { log.Printf("udp conn handler out to tcp crashed with err : %s \nstack: %s", err, string(debug.Stack())) } }() log.Printf("conn %d created , local: %s", connKey, srcAddr.String()) for { srcAddrFromConn, body, err := utils.ReadUDPPacket(&conn) if err == io.EOF || err == io.ErrUnexpectedEOF { //log.Printf("connection %d released", connKey) s.p.Remove(fmt.Sprintf("%d", connKey)) break } if err != nil { log.Printf("parse revecived udp packet fail, err: %s", err) continue } //log.Printf("udp packet revecived over parent , local:%s", srcAddrFromConn) _srcAddr := strings.Split(srcAddrFromConn, ":") if len(_srcAddr) != 2 { log.Printf("parse revecived udp packet fail, addr error : %s", srcAddrFromConn) continue } port, _ := strconv.Atoi(_srcAddr[1]) dstAddr := &net.UDPAddr{IP: net.ParseIP(_srcAddr[0]), Port: port} _, err = s.sc.UDPListener.WriteToUDP(body, dstAddr) if err != nil { log.Printf("udp response to local %s fail,ERR:%s", srcAddr, err) continue } //log.Printf("udp response to local %s success", srcAddr) } }() } //log.Printf("select conn %d , local: %s", connKey, srcAddr.String()) writer := bufio.NewWriter(conn) //fmt.Println(conn, writer) writer.Write(utils.UDPPacket(srcAddr.String(), packet)) err = writer.Flush() if err != nil { log.Printf("write udp packet to %s fail ,flush err:%s", *s.cfg.Parent, err) return } //log.Printf("write packet %v", packet) return } func (s *UDP) OutToUDP(packet []byte, localAddr, srcAddr *net.UDPAddr) (err error) { //log.Printf("udp packet revecived:%s,%v", srcAddr, packet) dstAddr, err := net.ResolveUDPAddr("udp", *s.cfg.Parent) if err != nil { log.Printf("resolve udp addr %s fail fail,ERR:%s", dstAddr.String(), err) return } clientSrcAddr := &net.UDPAddr{IP: net.IPv4zero, Port: 0} conn, err := net.DialUDP("udp", clientSrcAddr, dstAddr) if err != nil { log.Printf("connect to udp %s fail,ERR:%s", dstAddr.String(), err) return } conn.SetDeadline(time.Now().Add(time.Millisecond * time.Duration(*s.cfg.Timeout))) _, err = conn.Write(packet) if err != nil { log.Printf("send udp packet to %s fail,ERR:%s", dstAddr.String(), err) return } //log.Printf("send udp packet to %s success", dstAddr.String()) buf := make([]byte, 512) len, _, err := conn.ReadFromUDP(buf) if err != nil { log.Printf("read udp response from %s fail ,ERR:%s", dstAddr.String(), err) return } //log.Printf("revecived udp packet from %s , %v", dstAddr.String(), respBody) _, err = s.sc.UDPListener.WriteToUDP(buf[0:len], srcAddr) if err != nil { log.Printf("send udp response to cluster fail ,ERR:%s", err) return } //log.Printf("send udp response to cluster success ,from:%s", dstAddr.String()) return } func (s *UDP) InitOutConnPool() { if *s.cfg.ParentType == TYPE_TLS || *s.cfg.ParentType == TYPE_TCP { //dur int, isTLS bool, certBytes, keyBytes []byte, //parent string, timeout int, InitialCap int, MaxCap int s.outPool = utils.NewOutPool( *s.cfg.CheckParentInterval, *s.cfg.ParentType == TYPE_TLS, s.cfg.CertBytes, s.cfg.KeyBytes, *s.cfg.Parent, *s.cfg.Timeout, *s.cfg.PoolSize, *s.cfg.PoolSize*2, ) } }