browserSpider/program/tun/main.go

package main

import (
	"container/list"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"io"
	"jycdp"
	"log"
	"net"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"
)

const (
	tun_status_normal = iota
	tun_status_busy
	tun_status_idle
)

type (
	//Springboard 跳板
	Springboard struct {
		TcpServeAddr string `yaml:"address"`
		Cache        *list.List
		Proxy        *IpItem //代理地址
	}
	//Config
	Config struct {
		MonitorAddress         string `yaml:"monitor_address"`
		ConnectTimeout         int64  `yaml:"connect_timeout"`
		SpringboardesPortRange [2]int `yaml:"portrange"`
		PublicIpAddr           string `yaml:"public_ip_address"`
	}
	//流量监听 ，只监听请求端
	Monitor struct {
		s *Springboard
	}
	//
	MyMux struct{}
)

// Write 实现写接口
func (m *Monitor) Write(b []byte) (int, error) {
	if strings.Contains(string(b), "pleasechangemyip.com") {
		m.s.ChangeIp()
	}
	return len(b), nil
}

var (
	configFile         = flag.String("c", "./config.yaml", "配置文件")
	config     *Config = new(Config)
	status     []int
	tunLock    = new(sync.RWMutex)
)

// forward 流量转发
func (s *Springboard) forward(src, dest io.ReadWriteCloser) {
	defer func() {
		if src != nil {
			_ = src.Close()
		}
		if dest != nil {
			_ = dest.Close()
		}
	}()
	if src == nil || dest == nil {
		return
	}
	_, _ = io.Copy(io.MultiWriter(&Monitor{s}, src), dest)
}

// processClient 处理客户端请求
func (s *Springboard) ProcessClient(client net.Conn) {
	defer func() {
		if err := recover(); err != nil {
			fmt.Printf("{err %v}", err)
		}
	}()
	var targetConn net.Conn
	var err error
	for {
		//默认复用上一次对可用IP
		targetConn, err = net.DialTimeout("tcp",
			s.Proxy.Raw,
			time.Duration(config.ConnectTimeout)*time.Second)
		if err == nil {
			break
		}
		//TODO 移除有问题的链接
		//roulette.ClearBadProxy(s.Proxy.Index)
		//跳板主动切换IP
		s.Proxy, _ = roulette.Get()
	}
	s.Cache.PushBack([2]net.Conn{client, targetConn})
	go s.forward(client, targetConn)
	go s.forward(targetConn, client)
}

// StartServe 开启TCP服务
func (s *Springboard) StartServe() {
	s.Proxy, _ = roulette.Get()
	log.Println(s.TcpServeAddr, s.Proxy.Raw)
	conn, err := net.Listen("tcp", s.TcpServeAddr)
	if err != nil {
		log.Println(err.Error())
		return
	}
	log.Println("springboard listen:", s.TcpServeAddr)
	for {
		client, err := conn.Accept()
		if err != nil {
			log.Println("Listen failed: %v\n", err)
			continue
		}
		go s.ProcessClient(client)
	}
}

// ChangeIp 切换IP
func (s *Springboard) ChangeIp() {
	s.Proxy, _ = roulette.Get()
	fmt.Print("C")
	//fmt.Printf("(%s)", s.Proxy.Raw)
	for e := s.Cache.Front(); e != nil; e = e.Next() {
		connes, _ := e.Value.([2](net.Conn))
		if connes[0].Close() != nil {
			_ = connes[0].Close()
		}
		if connes[1].Close() != nil {
			_ = connes[0].Close()
		}
	}
}

// Apply 批量申请代理端口资源，用于启动docker，传入proxy_server
func Apply(length int) (ret []string, err error) {
	tunLock.Lock()
	defer tunLock.Unlock()
	ret = make([]string, length, length)
	pos, start := 0, config.SpringboardesPortRange[0]
	tmp := make([]int, length, length)
	for i, v := range status {
		if v == tun_status_idle {
			//status[i] = tun_status_busy
			tmp[pos] = i
			pos += 1
		}
		if pos == length {
			break
		}
	}
	if pos < length-1 {
		return nil, errors.New("没有足够的tun资源")
	}
	//改状态
	for i, v := range tmp {
		status[v] = tun_status_busy
		ret[i] = fmt.Sprintf("%s:%d", config.PublicIpAddr, start+v)
	}
	return
}

// Repay 归还释放tun资源
func Repay(address []string) {
	ports := make([]int, len(address), len(address))
	for i, v := range address {
		ports[i], _ = strconv.Atoi(strings.Split(v, ":")[1])
	}
	tunLock.Lock()
	defer tunLock.Unlock()
	start := config.SpringboardesPortRange[0]
	for _, v := range ports {
		status[v-start] = tun_status_idle
	}
}

// monitor
func monitor() {
	//申请tun端口
	http.HandleFunc("/apply", func(w http.ResponseWriter, r *http.Request) {
		lenStr := r.FormValue("len")
		if lenStr == "" {
			lenStr = "1"
		}
		length, _ := strconv.Atoi(lenStr)
		ret, err := Apply(length)
		if err != nil {
			w.WriteHeader(http.StatusInternalServerError)
			fmt.Fprint(w, "没有足够的tun资源，请减少申请数量")
		} else {
			json.NewEncoder(w).Encode(ret)
		}
	})
	http.HandleFunc("/repay", func(w http.ResponseWriter, r *http.Request) {
		data := struct {
			Data []string `json:"address"`
		}{}
		json.NewDecoder(r.Body).Decode(&data)
		Repay(data.Data)
		fmt.Fprint(w, "ok")
	})
	http.ListenAndServe(config.MonitorAddress, nil)
}

// init 初始化
func init() {
	flag.Parse()
	jycdp.LoadConfig(*configFile, config)
	length := config.SpringboardesPortRange[1] - config.SpringboardesPortRange[0] + 1
	status = make([]int, length, length)
	//全部设置为闲
	for i, _ := range status {
		status[i] = tun_status_idle
	}
	reloadProxies()
	go watch()
}

// main
func main() {
	//
	for i := config.SpringboardesPortRange[0]; i <= config.SpringboardesPortRange[1]; i++ {
		s := &Springboard{
			TcpServeAddr: fmt.Sprintf(":%d", i),
			Cache:        list.New(),
		}
		go s.StartServe()
	}
	monitor()
	//
}