transx/tunnel.go

package main

import (
	"bytes"
	"crypto/md5"
	"encoding/hex"
	"github.com/TransX/log"
	"github.com/TransX/tscipher"
	"net"
	"strconv"
	"sync/atomic"
	"time"
)

var seed int32

func init() {
	seed = 0
}

type Tunnel struct {
	id              string
	src             net.Conn
	dest            net.Conn
	cipherDirection string
}

func NewTunnel(src, dest net.Conn, cipherDirection string) *Tunnel {
	return &Tunnel{
		id:              tunnelID(),
		src:             src,
		dest:            dest,
		cipherDirection: cipherDirection,
	}
}

func (this *Tunnel) GetID(id string) string {
	return this.id
}

func (this *Tunnel) SetID(id string) { //rarely used
	this.id = id
}

//tunnel model : [ -->>server ---- client -->> ](this is a tunnel)
func (this *Tunnel) run() { //单向的，从src发送到dest
	src := this.src
	dest := this.dest
	cipherDirection := this.cipherDirection
	id := this.id
	defer func() {
		if r := recover(); r != nil {
			if src != nil {
				src.Close()
			}
			if dest != nil {
				dest.Close()
			}

		}
	}()
	cache := make([]byte, 1024*32) //128kB
	//构建Carrier
	// revCarrier := &tscipher.Carrier{
	// 	src,
	// 	tscipher.NewCipher("XOR"),
	// 	cache,
	// 	this.id,
	// }
	revCarrier := tscipher.NewCarrier(src, tscipher.NewCipher("XOR"), cache, this.id)
	// sendCarrier := &tscipher.Carrier{
	// 	dest,
	// 	tscipher.NewCipher("XOR"),
	// 	cache, //TODO:危险，cache的容量容易被不小心修改
	// 	this.id,
	// }
	sendCarrier := tscipher.NewCarrier(dest, tscipher.NewCipher("XOR"), cache, this.id)
	for {
		// log.Info("a loop Tunnel id %s", id)
		var nByte int
		// log.Debug("Encrypt Direction %s ID %s", cipherDirection, id)
		var err error
		if cipherDirection != "receive" {
			revCarrier.Cipher = nil
			// log.Debug("Read not crypted. Tunnel: %s", id)
			nByte, err = tscipher.RowReceiveData(revCarrier)
		} else {
			nByte, err = tscipher.ReceiveData(revCarrier)
		}
		if err != nil {
			log.Panic("Read panic. Tunnel id: %s. Remote Add: %s Local: %s. Err:%s", id, src.RemoteAddr().String(), src.LocalAddr().String(), err.Error())
		}
		log.Info("Reived %d bytes from %s. Tunnel: id %s", nByte, src.RemoteAddr().String(), id)
		if cipherDirection != "send" {
			sendCarrier.Cipher = nil
			// log.Debug("Write not crypted. Tunnel: %s", id)
		}
		// for {
		n, err := tscipher.SendData(sendCarrier, nByte)
		if err != nil {
			log.Panic("Write panic. ID: %s, Err: %s, Remote Add: %s", id, err, dest.RemoteAddr().String())
		}
		log.Info("Write %d bytes from %s to %s. Tunnel: %s . 18 bytes", n, dest.LocalAddr(), dest.RemoteAddr().String(), id, string(sendCarrier.Cache[:18]))
		// if n == 0 {
		// 	log.Info("Retry Write id %s", id)
		// 	continue
		// }
		// break
		// }

	}

}

func tunnelID() string {
	nowString := time.Now().String() + strconv.Itoa(int(seed))
	atomic.AddInt32(&seed, 1) //避免多线程情况下获得的种子相同
	md5Byte := md5.Sum(bytes.NewBufferString(nowString).Bytes())
	return hex.EncodeToString(md5Byte[:])
}