修订vmess代码，文档

proxy/vmess/anti_replay.go

@@ -5,11 +5,16 @@ import (
 	"time"
 )
 
-//authid 120 second anti replay.
-const authid_antiReplyDuration = time.Second * 120
+const (
+	//authid 120 seconds anti replay.
+	authid_antiReplyDuration = time.Second * 120
+
+	//在v2ray的代码中找到的。暂未找到对应的 “3分钟内不重复” 的文档。先照葫芦画瓢，到时候再追究。
+	sessionAntiReplayDuration = time.Minute * 3
+)
 
 /*
-	我们用自己的代码 实现 authid 防重放 机制. 不用v2ray的代码实现。
+	我们用map的方式 实现 authid 防重放 机制. 不用v2ray的代码实现。
 
 	v2ray中 “使用 两个filter，每隔120秒swap一次filter” 的方式，感觉不够严谨。第121～240秒的话，
 	实际上 第一个 filter 仍然会存储之前所有数据，而只是第二个filter被重置了，导致 第一个filter是防240秒重放。
@@ -17,26 +22,26 @@ const authid_antiReplyDuration = time.Second * 120
 	因为这样的话，实际上把时间按120秒分块了，如果一个id在 第 1秒被使用，然后在第122秒被重新使用，按v2ray的实现，
 	依然会被认定为重放攻击。
 
-	我们只要和 sessionHistory的方式一样，存储过期时间，然后定时清理 即可。
+	我们只要用 v2ray的 sessionHistory的方式，存储过期时间，然后定时清理 即可。
 */
-type anitReplayMachine struct {
+type authid_antiReplayMachine struct {
 	sync.RWMutex
-	antiReplyMap map[[16]byte]time.Time //key: authid, value: expireTime
+	authidMap map[[16]byte]time.Time //key: authid, value: expireTime
 
 	ticker   *time.Ticker
 	stopChan chan struct{}
 	closed   bool
 }
 
-func newAntiReplyMachine() *anitReplayMachine {
-	arm := &anitReplayMachine{
-		antiReplyMap: make(map[[16]byte]time.Time),
-		ticker:       time.NewTicker(authid_antiReplyDuration * 2),
-		stopChan:     make(chan struct{}),
+func newAuthIDAntiReplyMachine() *authid_antiReplayMachine {
+	arm := &authid_antiReplayMachine{
+		authidMap: make(map[[16]byte]time.Time),
+		ticker:    time.NewTicker(authid_antiReplyDuration * 2),
+		stopChan:  make(chan struct{}),
 	}
 
 	//定时清理过时数据，避免缓存无限增长
-	go func(a *anitReplayMachine) {
+	go func(a *authid_antiReplayMachine) {
 
 		for {
 			select {
@@ -44,9 +49,9 @@ func newAntiReplyMachine() *anitReplayMachine {
 				return
 			case now := <-a.ticker.C:
 				a.Lock()
-				for authid, expireTime := range a.antiReplyMap {
+				for authid, expireTime := range a.authidMap {
 					if expireTime.Before(now) {
-						delete(a.antiReplyMap, authid)
+						delete(a.authidMap, authid)
 					}
 				}
 				a.Unlock()
@@ -56,7 +61,7 @@ func newAntiReplyMachine() *anitReplayMachine {
 	return arm
 }
 
-func (arm *anitReplayMachine) stop() {
+func (arm *authid_antiReplayMachine) stop() {
 	arm.Lock()
 	defer arm.Unlock()
 
@@ -69,22 +74,22 @@ func (arm *anitReplayMachine) stop() {
 
 }
 
-func (arm *anitReplayMachine) check(authid [16]byte) (ok bool) {
+func (arm *authid_antiReplayMachine) check(authid [16]byte) (ok bool) {
 	now := time.Now()
 	arm.RLock()
-	expireTime, has := arm.antiReplyMap[authid]
+	expireTime, has := arm.authidMap[authid]
 	arm.RUnlock()
 
 	if !has {
 		arm.Lock()
-		arm.antiReplyMap[authid] = now.Add(authid_antiReplyDuration)
+		arm.authidMap[authid] = now.Add(authid_antiReplyDuration)
 		arm.Unlock()
 
 		return true
 	}
 	if expireTime.Before(now) {
 		arm.Lock()
-		arm.antiReplyMap[authid] = now.Add(authid_antiReplyDuration)
+		arm.authidMap[authid] = now.Add(authid_antiReplyDuration)
 		arm.Unlock()
 
 		return true
@@ -97,53 +102,86 @@ type sessionID struct {
 	key   [16]byte
 	nonce [16]byte
 }
-type sessionHistory struct {
+type session_antiReplayMachine struct {
 	sync.RWMutex
-	cache map[sessionID]time.Time
+	sessionMap map[sessionID]time.Time
+
+	ticker   *time.Ticker
+	stopChan chan struct{}
+	closed   bool
 }
 
-func NewSessionHistory() *sessionHistory {
-	h := &sessionHistory{}
+func newSessionAntiReplayMachine() *session_antiReplayMachine {
+	h := &session_antiReplayMachine{
+		ticker:   time.NewTicker(sessionAntiReplayDuration * 2),
+		stopChan: make(chan struct{}),
+	}
 	h.initCache()
 
+	//定时清理过时数据，避免缓存无限增长
+	go func(sh *session_antiReplayMachine) {
+
+		for {
+			select {
+			case <-sh.stopChan:
+				return
+			case now := <-sh.ticker.C:
+				sh.Lock()
+				sh.removeExpiredEntries(now)
+				sh.Unlock()
+			}
+		}
+	}(h)
+
 	return h
 }
 
-func (h *sessionHistory) initCache() {
-	h.cache = make(map[sessionID]time.Time, 128)
+func (sh *session_antiReplayMachine) stop() {
+	sh.Lock()
+	defer sh.Unlock()
+
+	if sh.closed {
+		return
+	}
+	sh.closed = true
+	close(sh.stopChan)
+	sh.ticker.Stop()
+
 }
 
-func (h *sessionHistory) addIfNotExits(session sessionID) bool {
+func (h *session_antiReplayMachine) initCache() {
+	h.sessionMap = make(map[sessionID]time.Time, 128)
+}
+
+func (h *session_antiReplayMachine) check(session sessionID) bool {
 	h.Lock()
 
 	now := time.Now()
 
-	h.removeExpiredEntries(now)
-
-	if expire, found := h.cache[session]; found && expire.After(now) {
+	if expire, found := h.sessionMap[session]; found && expire.After(now) {
 		h.Unlock()
 		return false
 	}
 
-	h.cache[session] = time.Now().Add(time.Minute * 3) //在v2ray的代码中找到的。暂未找到对应的 “3分钟内不重复” 的文档。
+	h.sessionMap[session] = now.Add(sessionAntiReplayDuration)
 	h.Unlock()
 
 	return true
 }
 
-func (h *sessionHistory) removeExpiredEntries(now time.Time) {
+func (h *session_antiReplayMachine) removeExpiredEntries(now time.Time) {
 
-	if len(h.cache) == 0 {
+	if len(h.sessionMap) == 0 {
 		return
 	}
 
-	for session, expire := range h.cache {
+	for session, expire := range h.sessionMap {
 		if expire.Before(now) {
-			delete(h.cache, session)
+			delete(h.sessionMap, session)
 		}
 	}
 
-	if len(h.cache) == 0 {
+	if len(h.sessionMap) == 0 {
 		h.initCache() //这是为了回收内存。
 	}
 

proxy/vmess/client.go

@@ -379,16 +379,17 @@ func (c *ClientConn) Write(b []byte) (n int, err error) {
 		}
 	}
 
-	n, err = c.dataWriter.Write(b)
-	if len(b) != 0 {
+	if len(b) > 0 {
+		n, err = c.dataWriter.Write(b)
 		close(switchChan)
 		c.vmessout = nil
 
+		if err != nil {
+			return
+		}
+		_, err = c.Conn.Write(outBuf.Bytes())
 	}
-	if err != nil {
-		return
-	}
-	_, err = c.Conn.Write(outBuf.Bytes())
+
 	return
 }
 

proxy/vmess/header.go

@@ -90,7 +90,7 @@ func generateCipherByV2rayUser(u utils.V2rayUser) (cipher.Block, error) {
 }
 
 //为0表示匹配成功, 如果不为0，则匹配失败；若为1，则CRC 校验失败（正常地匹配失败，不意味着被攻击）; 若为2，则表明校验成功 但是 时间差距超过 authID_timeMaxSecondGap 秒，如果为3，则表明遇到了重放攻击。
-func tryMatchAuthIDByBlock(now int64, block cipher.Block, encrypted_authID [16]byte, anitReplayMachine *anitReplayMachine) (failReason int) {
+func tryMatchAuthIDByBlock(now int64, block cipher.Block, encrypted_authID [16]byte, anitReplayMachine *authid_antiReplayMachine) (failReason int) {
 
 	var t int64
 	//var rand int32

proxy/vmess/server.go

@@ -34,7 +34,7 @@ type pair struct {
 	cipher.Block
 }
 
-func authUserByAuthPairList(bs []byte, authPairList []pair, anitReplayMachine *anitReplayMachine) (user utils.V2rayUser, err error) {
+func authUserByAuthPairList(bs []byte, authPairList []pair, anitReplayMachine *authid_antiReplayMachine) (user utils.V2rayUser, err error) {
 	now := time.Now().Unix()
 
 	var encrypted_authid [authid_len]byte
@@ -110,16 +110,15 @@ type Server struct {
 
 	authPairList []pair
 
-	sessionHistory *sessionHistory
-
-	anitReplayMachine *anitReplayMachine
+	authid_anitReplayMachine  *authid_antiReplayMachine
+	session_antiReplayMachine *session_antiReplayMachine
 }
 
 func NewServer() *Server {
 	s := &Server{
-		MultiUserMap:      utils.NewMultiUserMap(),
-		sessionHistory:    NewSessionHistory(),
-		anitReplayMachine: newAntiReplyMachine(),
+		MultiUserMap:              utils.NewMultiUserMap(),
+		authid_anitReplayMachine:  newAuthIDAntiReplyMachine(),
+		session_antiReplayMachine: newSessionAntiReplayMachine(),
 	}
 	s.SetUseUUIDStr_asKey()
 	return s
@@ -127,7 +126,8 @@ func NewServer() *Server {
 func (s *Server) Name() string { return Name }
 
 func (s *Server) Stop() {
-	s.anitReplayMachine.stop()
+	s.authid_anitReplayMachine.stop()
+	s.session_antiReplayMachine.stop()
 }
 
 func (s *Server) addUser(u utils.V2rayUser) {
@@ -160,7 +160,7 @@ func (s *Server) Handshake(underlay net.Conn) (tcpConn net.Conn, msgConn netLaye
 		returnErr = utils.NumErr{E: utils.ErrInvalidData, N: 1}
 		return
 	}
-	user, err := authUserByAuthPairList(data[:authid_len], s.authPairList, s.anitReplayMachine)
+	user, err := authUserByAuthPairList(data[:authid_len], s.authPairList, s.authid_anitReplayMachine)
 	if err != nil {
 
 		returnErr = err
@@ -176,6 +176,13 @@ func (s *Server) Handshake(underlay net.Conn) (tcpConn net.Conn, msgConn netLaye
 		returnErr = errorReason
 
 		if ce := utils.CanLogWarn("vmess openAEADHeader err"); ce != nil {
+			//看v2ray有一个 "drain"的用法，
+			//然而，我们这里是不需要drain的，区别在于，v2ray 不是一次性读取一大串数据，
+			// 而是用一个 reader 一点一点读，这就会产生一些可探测的问题，所以才要drain
+			// 而我们直接用 64K 的大buf 一下子读取整个客户端发来的整个数据， 没有读取长度的差别。
+
+			//不过 为了尊重v2ray的代码，也 以防 我的想法有错误，还是把这个情况陈列在这里，留作备用。
+
 			ce.Write(zap.Any("things", []any{errorReason, shouldDrain, bytesRead}))
 		}
 
@@ -211,7 +218,7 @@ func (s *Server) Handshake(underlay net.Conn) (tcpConn net.Conn, msgConn netLaye
 	sid.key = sc.reqBodyKey
 	sid.nonce = sc.reqBodyIV
 
-	if !s.sessionHistory.addIfNotExits(sid) {
+	if !s.session_antiReplayMachine.check(sid) {
 		returnErr = ErrReplaySessionAttack
 		return
 	}