Files

729 lines
16 KiB
Go

package server
import (
"bytes"
"cert"
"crypto/tls"
"encoding/json"
"fmt"
"hash/crc32"
"io"
"net"
"net/url"
"rakshasa/aes"
"rakshasa/common"
"runtime/debug"
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
)
var (
bufPool = &sync.Pool{
New: func() interface{} {
return &bytes.Buffer{}
},
}
closeChan = make(chan *bytes.Buffer, 1) //用于接收已关闭消息的黑洞chan
)
// 节点的连接,包含listen来的和主动connect的
type Conn struct {
closeTag int32
node *node
nodeaddr string
//key string
remoteAddr string
inChan chan func()
OutChan chan []byte
close chan string
isClient bool
nodeConn *tls.Conn
regResult chan RegMsg
}
type serverListen struct {
close int32
node *node
listen net.Listener
isSocks5 bool
socks5Replay []byte
replayid uint32
id uint32
connMap sync.Map
randkey []byte
}
type serverConnect struct {
close int32
id uint32
windowsSize int64
conn net.Conn
node *node
address string
write chan *bytes.Buffer
wait chan int
closeReason string
randkey []byte
}
// 中转与最终出口
func StartServer(addr string) error {
config := cert.Tlsconfig.Clone()
fmt.Println("start on ", addr)
l, err := tls.Listen("tcp", addr, config)
if err != nil {
return fmt.Errorf("server start fail %v", err)
}
currentNode.listen = l
go func() {
for {
conn, err := l.Accept()
if err != nil {
if err.(*net.OpError).Err == net.ErrClosed {
return
}
continue
}
//封装一个符合common.server接口的server
c := &Conn{
nodeConn: conn.(*tls.Conn),
remoteAddr: conn.RemoteAddr().String(),
}
connMap.Store(c.remoteAddr, conn)
go c.handlerNodeRead()
go c.handle()
}
}()
return nil
}
func init() {
go func() {
for b := range closeChan {
b.Reset()
bufPool.Put(b)
}
}()
}
func (conn *serverConnect) Close(reason string) {
if atomic.CompareAndSwapInt32(&conn.close, 0, 1) {
go func() {
if conn.conn != nil {
conn.conn.Close()
}
//fmt.Println(conn.fd, reason)
conn.node.connMap.Delete(conn.id)
conn.closeReason = reason
conn.node.listenMap.Range(func(key, value interface{}) bool {
value.(*serverListen).connMap.Delete(conn.id)
return true
})
if reason != remoteClose {
conn.node.Write(common.CMD_DELETE_CONNID, conn.id, nil)
}
select {
case conn.wait <- common.CONN_STATUS_CLOSE:
case <-time.After(time.Second * 10):
}
conn.write <- nil
conn.write = closeChan
}()
}
}
func (c *Conn) Close(reason string) {
c.close <- reason
}
func (conn *serverConnect) handTcpReceive() {
go func() {
for b := range conn.write {
if b == nil {
conn.write = closeChan
return
}
if _, err := conn.conn.Write(b.Bytes()); err != nil {
conn.Close(err.Error())
}
b.Reset()
bufPool.Put(b)
}
}()
var err error
var n int
defer func() {
if err != nil {
conn.Close(conn.address + " 读取出错" + err.Error())
} else {
conn.Close(conn.address + " read异常关闭")
}
}()
buf := make([]byte, common.MAX_PLAINTEXT)
for conn.close == 0 {
conn.conn.SetReadDeadline(time.Now().Add(common.WRITE_DEADLINE))
n, err = conn.conn.Read(buf)
if err != nil {
if atomic.LoadInt32(&conn.close) == 0 {
if e := err.Error(); !strings.Contains(e, ": i/o timeout") {
return
}
continue
} else {
return
}
}
data := make([]byte, n)
copy(data, buf)
if common.Debug {
fmt.Println("发送", crc32.ChecksumIEEE(data), n)
}
conn.node.Write(common.CMD_CONN_MSG, conn.id, data)
atomic.AddInt64(&conn.windowsSize, -1*int64(n))
for atomic.LoadInt64(&conn.windowsSize) <= 0 && conn.close == 0 {
select {
case flag := <-conn.wait:
if flag == common.CONN_STATUS_CLOSE {
return
}
case <-time.After(time.Second):
}
}
}
}
func (conn *serverConnect) Write(data []byte) {
data = data[1:]
windows_update_size := int64(data[0]) | int64(data[1])<<8 | int64(data[2])<<16 | int64(data[3])<<24 | int64(data[4])<<32 | int64(data[5])<<40 | int64(data[6])<<48 | int64(data[7])<<56
if windows_update_size != 0 {
old := atomic.AddInt64(&conn.windowsSize, windows_update_size) - windows_update_size
if old < 0 {
go func() {
select {
case conn.wait <- common.CONN_STATUS_OK:
case <-time.After(time.Second):
}
}()
}
}
if common.Debug {
fmt.Println("收到", crc32.ChecksumIEEE(data[8:]), len(data[8:]))
}
b := bufPool.Get().(*bytes.Buffer)
b.Reset()
b.Write(data[8:])
conn.write <- b
}
func (conn *serverConnect) handUdpReceive() {
var err error
var n int
defer func() {
if err != nil {
conn.Close(conn.address + " 网站读取出错" + err.Error())
} else {
conn.Close(conn.address + " read异常关闭")
}
}()
buf := make([]byte, common.MAX_PLAINTEXT)
for conn.close == 0 {
conn.conn.SetReadDeadline(time.Now().Add(common.WRITE_DEADLINE))
n, err = conn.conn.Read(buf)
if err != nil {
if atomic.LoadInt32(&conn.close) == 0 {
if e := err.Error(); !strings.Contains(e, ": i/o timeout") {
return
}
continue
} else {
return
}
}
b := make([]byte, n)
copy(b, buf)
conn.node.Write(common.CMD_CONN_UDP_MSG, conn.id, b)
}
}
func (conn *serverConnect) doConnectTcp(network common.NetWork, addr string) {
netconn, err := net.DialTimeout("tcp", addr, time.Second*30)
if err != nil {
buf := make([]byte, 2)
buf[0] = byte(network)
buf[1] = 0
conn.node.Write(common.CMD_CONNECT_BYIDADDR_RESULT, conn.id, append(conn.randkey, buf...))
conn.Close("fd拨号失败")
return
} else {
buf := make([]byte, 2)
buf[0] = byte(network)
buf[1] = 1
conn.node.Write(common.CMD_CONNECT_BYIDADDR_RESULT, conn.id, append(conn.randkey, buf...))
if conn.close == 0 {
conn.conn = netconn
go conn.handTcpReceive()
}
}
}
func (conn *serverConnect) doConnectTcpWithHttpProxy(network common.NetWork, addr string) {
writeResult := func(res bool) {
buf := make([]byte, 2)
buf[0] = byte(network)
buf[1] = 0
if res {
buf[1] = 1
}
conn.node.Write(common.CMD_CONNECT_BYIDADDR_RESULT, conn.id, append(conn.randkey, buf...))
}
if i := strings.IndexByte(addr, 32); i > -1 {
cfg, err := common.ParseAddr(addr[i+1:])
if err != nil {
writeResult(false)
conn.Close("地址解析失败")
return
}
netconn, err := net.DialTimeout("tcp", cfg.Addr(), time.Second*2)
if err != nil {
writeResult(false)
conn.Close("fd拨号失败")
return
} else {
netconn.SetDeadline(time.Now().Add(time.Second * 30))
netconn.SetWriteDeadline(time.Now().Add(time.Second * 30))
switch cfg.Scheam() {
case "", "http://":
//请求代理
data := fmt.Sprintf("CONNECT %s HTTP/1.1\r\nHost: %s\r\nProxy-Connection: keep-alive\r\nUser-Agent: Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.198 Safari/537.36\r\n", addr[:i], addr[:i])
if cfg.GetHttpAuthorizationHeader() != "" {
data += cfg.GetHttpAuthorizationHeader() + "\r\n\r\n"
} else {
data += "\r\n"
}
_, err = netconn.Write([]byte(data))
if err != nil {
writeResult(false)
conn.Close("http代理发送消息失败")
return
}
var resdata []byte
var result [8192]byte
var req = &http1request{}
for {
n, err := netconn.Read(result[:])
if err != nil {
writeResult(false)
conn.Close("读取http代理结果失败")
return
}
resdata = append(resdata, result[:n]...)
l, _, err := parsereq(req, resdata)
if err != nil {
return
} else if l > 0 {
break
}
}
if req.Status == "200 Connection established" {
writeResult(true)
if conn.close == 0 {
conn.conn = netconn
go conn.handTcpReceive()
}
} else {
writeResult(false)
conn.Close("http代理连接失败")
}
writeResult(true)
case "socks5://":
_, err = netconn.Write([]byte{5, 1, 2})
if err != nil {
writeResult(false)
conn.Close("socks5代理发送消息失败")
return
}
var result [8192]byte
n, err := netconn.Read(result[:])
if err != nil {
writeResult(false)
conn.Close("读取socks5数据出错")
return
}
if string(result[:n]) == string([]byte{5, 2}) { //需要认证
user, password := cfg.User(), cfg.Password()
if user == "" && password == "" {
writeResult(false)
conn.Close("socks5需要验证")
return
}
data := make([]byte, (3 + len(user) + len(password)))
data[0] = 5
data[1] = byte(len(user))
copy(data[2:], user)
data[2+len(user)] = byte(len(password))
copy(data[3+len(user):], password)
netconn.Write(data)
n, err = netconn.Read(result[:])
if err != nil || string(result[:n]) != string([]byte{5, 0}) {
writeResult(false)
conn.Close("密码校验不通过")
return
}
}
data := []byte{5, 1, 0, 1}
if u, err := url.ParseRequestURI(addr[:i]); err == nil {
data[3] = 3
data = append(data, byte(len(u.Scheme)))
data = append(data, u.Scheme...)
p, _ := strconv.Atoi(u.Opaque)
port := []byte{byte(p >> 8), byte(p)}
data = append(data, port...)
} else if tcp4, err := net.ResolveTCPAddr("tcp4", addr[:i]); err == nil {
data[3] = 1
data = append(data, tcp4.IP.String()...)
port := []byte{byte(tcp4.Port), byte(tcp4.Port >> 8)}
data = append(data, port...)
} else if tcp6, err := net.ResolveTCPAddr("tcp6", addr[:i]); err == nil {
data[3] = 4
data = append(data, tcp6.IP.String()...)
port := []byte{byte(tcp6.Port), byte(tcp6.Port >> 8)}
data = append(data, port...)
}
netconn.Write(data)
n, _ = netconn.Read(result[:])
if n >= 2 && string(result[:2]) == string([]byte{5, 0}) {
writeResult(true)
if conn.close == 0 {
conn.conn = netconn
go conn.handTcpReceive()
}
} else {
writeResult(false)
conn.Close("socks5连接失败")
}
default:
writeResult(false)
conn.Close("不支持的代理协议")
}
}
} else {
writeResult(false)
conn.Close("无法获取代理地址")
}
}
func (conn *serverConnect) doHandleUdp() {
for b := range conn.write {
if b == nil {
conn.write = closeChan
return
}
conn.conn.Write(b.Bytes())
b.Reset()
bufPool.Put(b)
}
}
var broadcastMap sync.Map //广播帧防止重复处理
func (c *Conn) handlerNodeRead() {
var err error
defer func() {
c.nodeConn.Close()
c.Close("read错误" + err.Error())
}()
lengbuf := make([]byte, 2)
for {
_, err = io.ReadFull(c.nodeConn, lengbuf)
if err != nil {
if strings.Contains(err.Error(), "i/o timeout") {
continue
}
return
}
buf := make([]byte, int(lengbuf[0])+int(lengbuf[1])<<8)
_, err = io.ReadFull(c.nodeConn, buf)
b := aes.AesCtrDecrypt(buf)
msg := common.UnmarshalMsg(b)
if common.Debug {
fmt.Println("fromto", msg.From, msg.To, common.CmdToName[msg.CmdOpteion], int(lengbuf[0])+int(lengbuf[1])<<8)
}
if msg.To == common.NoneUUID.String() && c.node == nil {
c.inChan <- func() {
newNode := &node{
conn: c,
}
newNode.do(msg)
}
} else if msg.To == currentNode.uuid {
v, ok := nodeMap.Load(msg.From)
if ok && v.(*node).port != 0 {
c.inChan <- func() {
v.(*node).do(msg)
}
} else {
if !ok {
newNode := &node{
uuid: msg.From,
conn: c,
waitMsg: []*common.Msg{msg},
}
result := make(chan interface{}, 1)
id := newNode.storeQuery(result)
nodeMap.Store(msg.From, newNode)
newNode.Write(common.CMD_GET_CURRENT_NODE, id, []byte{1}) //获取丢失节点的信息
go func() {
defer func() {
if err := recover(); err != nil {
fmt.Println(err)
debug.PrintStack()
}
newNode.deleteQuery(id)
}()
select {
case res := <-result:
if res == nil {
for _, m := range newNode.waitMsg {
c.inChan <- func() {
newNode.do(m)
}
}
}
case <-time.After(common.CMD_TIMEOUT):
newNode.Close("超时")
}
}()
} else {
if msg.CmdOpteion == common.CMD_GET_CURRENT_NODE_RESULT {
n := v.(*node)
var res chan interface{}
if _v, ok := n.loadQuery(msg.CmdId); !ok {
return
} else {
res = _v
}
var nmsg nodeInfo
err = json.Unmarshal(msg.CmdData, &nmsg)
if err != nil {
res <- err
return
}
n.hostName = cert.RSADecrypterStr(nmsg.HostName)
n.uuid = cert.RSADecrypterStr(nmsg.UUID)
if n.port, err = strconv.Atoi(cert.RSADecrypterStr(nmsg.Port)); err != nil {
n.port = -1
}
n.mainIp = cert.RSADecrypterStr(nmsg.MainIp)
n.goos = cert.RSADecrypterStr(nmsg.Goos)
res <- nil
} else {
v.(*node).waitMsg = append(v.(*node).waitMsg, msg)
}
}
}
} else {
key := msg.From + "_" + strconv.Itoa(int(msg.MsgId))
msg.Ttl++
if _, ok := broadcastMap.LoadOrStore(key, struct{}{}); !ok {
if msg.From != currentNode.uuid && msg.To == common.BroadcastUUID.String() && msg.Ttl < 250 { //广播
go allNodesDo(func(_n *node) (bool, error) {
if _n.uuid != currentNode.uuid {
_n.WriteMsg(msg)
}
return true, nil
})
if common.Debug {
fmt.Println("广播do")
}
newNode := &node{
conn: c,
}
c.inChan <- func() {
newNode.do(msg)
}
} else {
c.WriteToUUID(msg)
}
time.AfterFunc(time.Hour, func() {
broadcastMap.Delete(key)
})
}
}
}
}
func (c *Conn) handle() {
c.OutChan = make(chan []byte, 64)
c.inChan = make(chan func())
c.close = make(chan string, 999)
go func() {
for {
select {
case f := <-c.inChan:
f()
//c.do(b)
case b := <-c.OutChan:
if c.closeTag == 0 {
c.tlsWrite(b)
//var err error
for i := 0; i < len(c.OutChan); i++ {
c.tlsWrite(<-c.OutChan)
}
}
case reason := <-c.close:
c.OutChan = upNodeWrite
if c.node != nil && c.node.nextPingTime > time.Now().Unix()+5 {
c.node.ping(0)
c.node.nextPingTime = time.Now().Unix() + 5
}
connMap.Delete(c.remoteAddr)
if atomic.CompareAndSwapInt32(&c.closeTag, 0, 1) {
if common.Debug {
fmt.Println(c.nodeConn.RemoteAddr().String(), "关闭原因", reason)
}
if c.nodeConn != nil {
if common.Debug {
fmt.Println("執行close1")
}
c.nodeConn.Close()
}
if c.node != nil {
c.node.Close(reason)
//移除上游连接
for i := len(upLevelNode) - 1; i >= 0; i-- {
n := upLevelNode[i]
if n.uuid == c.node.uuid {
upLevelNode = append(upLevelNode[:i], upLevelNode[i+1:]...)
}
}
if common.Debug {
fmt.Println("upLevelNode", len(upLevelNode))
}
}
}
return
}
}
}()
}
func (c *Conn) reg() error {
var err error
reg := &RegMsg{
UUID: currentNode.uuid,
MainIp: cert.RSAEncrypterStr(currentNode.mainIp),
Port: cert.RSAEncrypterStr(strconv.Itoa(currentNode.port)),
Goos: cert.RSAEncrypterStr(currentNode.goos),
Hostname: cert.RSAEncrypterStr(currentNode.hostName),
}
regb, _ := json.Marshal(reg)
msg := common.Msg{
From: currentNode.uuid,
To: common.NoneUUID.String(),
CmdOpteion: common.CMD_REG,
CmdData: regb,
}
if err = c.tlsWrite(msg.Marshal()); err != nil {
return err
}
go c.handlerNodeRead()
return nil
}
func (c *Conn) WriteToUUID(msg *common.Msg) {
if n, ok := nodeMap.Load(msg.To); ok {
n.(*node).WriteMsg(msg)
}
}
func (c *Conn) Write(b []byte) {
c.OutChan <- b
}
func (c *Conn) tlsWrite(b []byte) error {
c.nodeConn.SetWriteDeadline(time.Now().Add(common.WRITE_DEADLINE))
n, err := c.nodeConn.Write(b)
if common.Debug {
if c.node != nil {
fmt.Println("writeto", c.node.uuid, n)
} else {
fmt.Println("writeto", common.NoneUUID, n)
}
}
if err != nil {
c.Close("Write " + err.Error())
upNodeWrite <- b
}
return err
}