func (nego *d5SNegotiation) verifyThenDHExchange(conn net.Conn, credBuf []byte) (key []byte) {
	userIdentity, err := RSADecrypt(credBuf, nego.RSAKeys.priv)
	ThrowErr(err)
	clientIdentity := string(userIdentity)
	if log.V(2) {
		log.Infoln("Auth clientIdentity:", SubstringBefore(clientIdentity, IDENTITY_SEP), "***")
	}
	allow, ex := nego.AuthSys.Authenticate(userIdentity)
	cDHPub, err := ReadFullByLen(2, conn)
	if !allow { // invalid user indentity
		log.Warningf("Auth %s failed: %v\n", clientIdentity, ex)
		conn.Write([]byte{0, 1, 0xff})
		panic(ex)
	}
	nego.clientIdentity = clientIdentity
	key = takeSharedKey(nego.dhKeys, cDHPub)
	//	if log.V(5) {
	//		dumpHex("Sharedkey", key)
	//	}
	buf := new(bytes.Buffer)
	buf.Write(nego.dhKeys.pubLen)
	buf.Write(nego.dhKeys.pub)
	_, err = buf.WriteTo(conn)
	return
}

func warning(e interface{}) interface{} {
	if err, y := e.(error); y {
		// 0.9.x to 0.10.x config error
		if strings.HasPrefix(err.Error(), "Unrecognized directives") {
			log.Warningf("Please read %s/wiki to learn more.", project_url)
		}
	}
	return e
}

func (t *Session) tokensHandle(args []byte) {
	var cmd = args[0]
	switch cmd {
	case FRAME_ACTION_TOKEN_REQUEST:
		tokens := t.mgr.createTokens(t, GENERATE_TOKEN_NUM)
		tokens[0] = FRAME_ACTION_TOKEN_REPLY
		t.mux.bestSend(tokens, "replyTokens")
	default:
		log.Warningf("Unrecognized command=%x packet=[% x]\n", cmd, args)
	}
}

func (p *multiplexer) connectToDest(frm *frame, key string, tun *Conn) {
	defer func() {
		ex.CatchException(recover())
	}()
	var (
		dstConn net.Conn
		err     error
		target  = string(frm.data)
		denied  = false
	)
	if p.filter != nil {
		denied = p.filter.Filter(target)
	}
	if !denied {
		dstConn, err = net.DialTimeout("tcp", target, GENERAL_SO_TIMEOUT)
	}
	frm.length = 0
	if err != nil || denied {
		p.router.removePreRegistered(key)
		if denied {
			frm.action = FRAME_ACTION_OPEN_DENIED
			log.Warningf("Denied request [%s] for %s\n", target, key)
		} else {
			frm.action = FRAME_ACTION_OPEN_N
			log.Warningf("Cannot connect to [%s] for %s error: %s\n", target, key, err)
		}
		tunWrite2(tun, frm)
	} else {
		edge := p.router.register(key, target, tun, dstConn, false) // write edge
		if log.V(1) {
			log.Infoln("OPEN", target, "for", key)
		}
		dstConn.SetReadDeadline(ZERO_TIME)
		frm.action = FRAME_ACTION_OPEN_Y
		if tunWrite2(tun, frm) == nil {
			p.relay(edge, tun, frm.sid) // read edge
		} else { // send open_y failed
			SafeClose(tun)
		}
	}
}

func tunWrite2(tun *Conn, frm *frame) (err error) {
	err = tun.SetWriteDeadline(time.Now().Add(GENERAL_SO_TIMEOUT * 2))
	if err != nil {
		return
	}
	var nr, nw int
	nr = int(frm.length) + FRAME_HEADER_LEN
	nw, err = tun.Write(frm.toNewBuffer())
	if nr != nw || err != nil {
		log.Warningf("Write tun(%s) error(%v) when sending %s\n", tun.sign(), err, frm)
		SafeClose(tun)
		return
	}
	return nil
}

func tunWrite1(tun *Conn, buf []byte) (err error) {
	err = tun.SetWriteDeadline(time.Now().Add(GENERAL_SO_TIMEOUT * 2))
	if err != nil {
		return
	}
	var nr, nw int
	nr = len(buf)
	nw, err = tun.Write(buf)
	if nr != nw || err != nil {
		log.Warningf("Write tun(%s) error(%v) when sending buf.len=%d\n", tun.sign(), err, nr)
		SafeClose(tun)
		return
	}
	return nil
}

func sendFrame(frm *frame) (werr bool) {
	dst := frm.conn.conn
	if log.V(5) {
		log.Infoln("SEND queue", frm)
	}
	dst.SetWriteDeadline(time.Now().Add(GENERAL_SO_TIMEOUT))
	nw, ew := dst.Write(frm.data)
	if nw == int(frm.length) && ew == nil {
		return
	}
	werr = true
	// an error occured
	log.Warningf("Write edge(%s) error(%v). %s\n", frm.conn.dest, ew, frm)
	return
}

func (c *Client) saveTokens(data []byte) {
	var tokens []byte
	switch data[0] {
	case FRAME_ACTION_TOKEN_REQUEST:
		log.Warningf("unexpected token request")
		return
	case FRAME_ACTION_TOKEN_REPLY:
		tokens = data[1:]
	}
	c.lock.Lock()
	defer c.lock.Unlock()
	c.token = append(c.token, tokens...)
	atomic.CompareAndSwapInt32(&c.State, CLT_PENDING, CLT_WORKING)
	c.pendingTK.notifyAll()
	if log.V(3) {
		log.Infof("Recv tokens=%d pool=%d\n", len(tokens)/TKSZ, len(c.token)/TKSZ)
	}
}

func (c *Client) saveTokens(data []byte) {
	var tokens []byte
	switch data[0] {
	case FRAME_ACTION_TOKEN_REQUEST:
		log.Warningf("unexpected token request")
		return
	case FRAME_ACTION_TOKEN_REPLY:
		tokens = data[1:]
	}
	c.lock.Lock()
	c.token = append(c.token, tokens...)
	c.lock.Unlock()
	// wakeup waiting
	c.pendingTK.notifyAll()
	if log.V(3) {
		log.Infof("Recv tokens=%d pool=%d\n", len(tokens)/TKSZ, len(c.token)/TKSZ)
	}
}

// frame writer
func frameWriteBuffer(tun *Conn, origin []byte) (err error) {
	// default timeout is 10s
	err = tun.SetWriteDeadline(time.Now().Add(WRITE_TUN_TIMEOUT))
	if err == nil {
		var nw int
		buf := frameTransform(origin)
		nw, err = tun.Write(buf)
		if nw != len(buf) || err != nil {
			idleLastR := time.Now().UnixNano() - tun.priority.last
			if IsTimeout(err) && idleLastR < int64(WRITE_TUN_TIMEOUT) {
				err = nil
			} else {
				log.Warningf("Write tun (%s) error (%v) buf.len=%d\n", tun.sign(), err, len(buf))
				SafeClose(tun)
			}
		}
	}
	return
}

func (n *dbcSerNego) verifyThenDHExchange(conn net.Conn) (key []byte) {
	// client identity segment
	setRTimeout(conn)
	credBuf, err := ReadFullByLen(2, conn)
	ThrowErr(err)

	user, passwd, err := n.idBlockDeserialize(credBuf)
	ThrowErr(err)

	if log.V(1) {
		log.Infoln("Auth client", user)
	}

	allow, err := n.AuthSys.Authenticate(user, passwd)
	if allow {
		n.clientIdentity = user
	} else { // client denied
		log.Warningf("Auth %s:%s failed: %v\n", user, passwd, err)
		// reply failed msg
		conn.Write([]byte{0, 1, AUTH_FAILED})
		panic(err)
	}

	// read client RH-pub
	setRTimeout(conn)
	bobPub, err := ReadFullByLen(2, conn)
	ThrowErr(err)
	key, err = n.dhKey.ComputeKey(bobPub)
	ThrowErr(err)

	// send my RH-pub
	myPub := n.dhKey.ExportPubKey()
	buf := make([]byte, len(myPub)+2)
	binary.BigEndian.PutUint16(buf, uint16(len(myPub)))
	copy(buf[2:], myPub)

	setWTimeout(conn)
	_, err = conn.Write(buf)
	ThrowErr(err)
	return
}

func (n *dbcSerNego) negotiate(hConn *hashedConn, tcPool []uint64) (session *Session, err error) {
	var (
		nr  int
		buf = make([]byte, DP_P2I)
	)

	setRTimeout(hConn)
	nr, err = hConn.Read(buf)

	if nr == len(buf) {

		nr = 0 // reset nr
		ok, stype, len2 := verifyDbcHello(buf, n.sharedKey, tcPool)

		if ok {
			if len2 > 0 {
				setRTimeout(hConn)
				nr, err = io.ReadFull(hConn, buf[:len2])
			}

			if nr == int(len2) && err == nil {
				switch stype {
				case TYPE_NEW:
					return n.handshakeSession(hConn)
				case TYPE_DAT:
					return n.dataSession(hConn)
				}
			}
		}

	} // then may be a prober

	if err == nil {
		err = UNRECOGNIZED_REQ
	}

	// threats OR overlarge time error
	// We could use this log to block threats origin by external tools such as fail2ban.
	log.Warningf("Unrecognized Request from=%s len=%d\n", n.clientAddr, nr)
	return nil, err
}

func (n *dbcCltNego) validateAndGetTokens(hConn *hashedConn, t *tunParams) {
	setRTimeout(hConn)
	buf, err := ReadFullByLen(2, hConn)
	ThrowErr(err)

	// compare version with remote
	myVer := VERSION
	rVer := binary.BigEndian.Uint32(buf)
	ofs := 4
	if rVer > myVer {
		rVerStr := fmt.Sprintf("%d.%d.%04d", rVer>>24, (rVer>>16)&0xFF, rVer&0xFFFF)
		myVer >>= 16
		rVer >>= 16
		if myVer == rVer {
			log.Warningf("Caution !!! Please upgrade to new version, remote is v%s\n", rVerStr)
		} else {
			ThrowErr(INCOMPATIBLE_VERSION.Apply(rVerStr))
			// return
		}
	}

	// check ibHash
	_ibHash := buf[ofs : ofs+20]
	ofs += 20
	if !bytes.Equal(n.ibHash, _ibHash) {
		// S->C is polluted.
		ThrowErr(INCONSISTENT_HASH.Apply("MitM attack"))
	}

	// parse params
	t.deserialize(buf, ofs)

	if log.V(3) {
		log.Infof("Received tokens size=%d\n", len(t.token)/TKSZ)
	}

	// validated or throws
	verifyHash(hConn, false)
}

func (nego *d5CNegotiation) validateAndGetTokens(sconn *hashedConn, t *tunParams) {
	buf, err := ReadFullByLen(2, sconn)
	ThrowErr(err)
	tVer := VERSION
	oVer := binary.BigEndian.Uint32(buf)
	if oVer > tVer {
		oVerStr := fmt.Sprintf("%d.%d.%04d", oVer>>24, (oVer>>16)&0xFF, oVer&0xFFFF)
		tVer >>= 16
		oVer >>= 16
		if tVer == oVer {
			log.Warningf("Caution !!! Please upgrade to new version, remote is v%s\n", oVerStr)
		} else {
			err = INCOMPATIBLE_VERSION.Apply(oVerStr)
		}
		ThrowErr(err)
	}
	ofs := 4
	ofs += 2
	t.dtInterval = int(binary.BigEndian.Uint16(buf[ofs:]))
	ofs += 2
	t.tunQty = int(buf[ofs])
	t.token = buf[TUN_PARAMS_LEN:]
	if log.V(3) {
		n := len(buf) - TUN_PARAMS_LEN
		log.Infof("Received tokens count=%d\n", n/TKSZ)
	}
	rHash := sconn.RHashSum()
	wHash := sconn.WHashSum()
	_, err = sconn.Write(rHash)
	ThrowErr(err)
	oHash := make([]byte, TKSZ)
	_, err = sconn.Read(oHash)
	if !bytes.Equal(wHash, oHash) {
		log.Errorln("Server hash/r is inconsistence with the client/w")
		log.Errorf("rHash: [% x] wHash: [% x]\n", rHash, wHash)
		log.Errorf("oHash: [% x]\n", oHash)
		ThrowErr(INCONSISTENT_HASH)
	}
}

func (nego *d5SNegotiation) negotiate(hConn *hashedConn) (session *Session, err error) {
	setSoTimeout(hConn)
	nego.clientAddr = hConn.RemoteAddr().String()
	var (
		nr  int
		buf = make([]byte, DMLEN1)
	)
	nr, err = hConn.Read(buf)
	if err != nil {
		return nil, err
	}

	if nr == DMLEN2 &&
		d5SumValid(buf[TKSZ-2], buf[TKSZ]) && d5SumValid(buf[TKSZ-1], buf[TKSZ+1]) {
		return nego.dataSession(hConn, buf)
	}
	if nr == DMLEN1 &&
		d5SumValid(buf[0xd5], buf[0xff]) {
		return nego.handshakeSession(hConn, buf)
	}
	log.Warningf("Unrecognized Request from=%s len=%d\n", nego.clientAddr, nr)
	return nil, NEGOTIATION_FAILED
}

//.........这里部分代码省略.........
				frm.data = frm.data[:frm.length]
			}
		}
		if er != nil {
			// shutdown
			if atomic.LoadInt32(&p.status) < 0 {
				time.Sleep(time.Second)
				return
			}
			switch idle.consumeError(er) {
			case ERR_NEW_PING:
				if idle.ping(tun) == nil {
					continue
				}
			case ERR_PING_TIMEOUT:
				if log.V(1) {
					log.Errorln("Peer was unresponsive then close", tun.identifier)
				}
			default:
				if log.V(1) {
					log.Errorln("Error", tun.identifier, er)
				}
			}
			// abandon this connection
			return
		}
		key = sessionKey(tun, frm.sid)

		switch frm.action {
		case FRAME_ACTION_CLOSE_W:
			if edge, _ := router.getRegistered(key); edge != nil {
				edge.bitwiseCompareAndSet(TCP_CLOSE_W)
				edge.deliver(frm)
			}
		case FRAME_ACTION_CLOSE_R:
			if edge, _ := router.getRegistered(key); edge != nil {
				edge.bitwiseCompareAndSet(TCP_CLOSE_R)
				closeR(edge.conn)
			}
		case FRAME_ACTION_DATA:
			edge, pre := router.getRegistered(key)
			if edge != nil {
				edge.deliver(frm)
			} else if pre {
				router.preDeliver(key, frm)
			} else {
				if log.V(2) {
					log.Warningln("peer send data to an unexisted socket.", key, frm)
				}
				// trigger sending close to notice peer.
				pack(header, FRAME_ACTION_CLOSE_R, frm.sid, nil)
				if frameWriteBuffer(tun, header) != nil {
					return
				}
			}
		case FRAME_ACTION_OPEN:
			router.preRegister(key)
			go p.connectToDest(frm, key, tun)
		case FRAME_ACTION_OPEN_N, FRAME_ACTION_OPEN_Y, FRAME_ACTION_OPEN_DENIED:
			edge, _ := router.getRegistered(key)
			if edge != nil {
				if log.V(4) {
					log.Infoln(p.role, "recv OPEN_x", frm)
				}
				edge.ready <- frm.action
				close(edge.ready)
			} else {
				if log.V(2) {
					log.Warningln("peer send OPEN_x to an unexisted socket.", key, frm)
				}
			}
		case FRAME_ACTION_PING:
			if idle.pong(tun) == nil {
				atomic.AddInt32(&p.pingCnt, 1)
			} else { // reply pong failed
				return
			}
		case FRAME_ACTION_PONG:
			if idle.verify() {
				if p.isClient && idle.lastPing > 0 {
					sRtt, devRtt := idle.updateRtt()
					if DEBUG {
						log.Infof("sRtt=%d devRtt=%d", sRtt, devRtt)
					}
					if devRtt+(sRtt>>2) > sRtt {
						// restart ???
						log.Warningf("Unstable network sRtt=%d devRtt=%d", sRtt, devRtt)
					}
				}
			} else {
				log.Warningln("Incorrect action_pong received")
			}
		case FRAME_ACTION_TOKENS:
			handler(evt_tokens, frm.data)
		default:
			log.Errorln(p.role, "Unrecognized", frm)
		}
		tun.Update()
	}
}