package node import ( "fmt" "github.com/Yuzuki616/V2bX/api/panel" "github.com/Yuzuki616/V2bX/conf" "github.com/Yuzuki616/V2bX/core" "github.com/Yuzuki616/V2bX/core/app/dispatcher" "github.com/Yuzuki616/V2bX/node/legoCmd" "github.com/go-resty/resty/v2" "github.com/goccy/go-json" "github.com/xtls/xray-core/common/protocol" "github.com/xtls/xray-core/common/task" "log" "math" "reflect" "runtime" "time" ) type Node struct { server *core.Core config *conf.ControllerConfig clientInfo panel.ClientInfo apiClient panel.Panel nodeInfo *panel.NodeInfo Tag string userList []panel.UserInfo nodeInfoMonitorPeriodic *task.Periodic userReportPeriodic *task.Periodic onlineIpReportPeriodic *task.Periodic } // New return a Node service with default parameters. func New(server *core.Core, api panel.Panel, config *conf.ControllerConfig) *Node { controller := &Node{ server: server, config: config, apiClient: api, } return controller } // Start implement the Start() function of the service interface func (c *Node) Start() error { c.clientInfo = c.apiClient.Describe() // First fetch Node Info newNodeInfo, err := c.apiClient.GetNodeInfo() if err != nil { return err } c.nodeInfo = newNodeInfo c.Tag = c.buildNodeTag() // Add new tag err = c.addNewTag(newNodeInfo) if err != nil { log.Panic(err) return err } // Update user userInfo, err := c.apiClient.GetUserList() if err != nil { return err } err = c.addNewUser(userInfo, newNodeInfo) if err != nil { return err } //sync controller userList c.userList = userInfo if err := c.server.AddInboundLimiter(c.Tag, c.nodeInfo, userInfo); err != nil { log.Print(err) } // Add Rule Manager if !c.config.DisableGetRule { if ruleList, protocolRule, err := c.apiClient.GetNodeRule(); err != nil { log.Printf("Get rule list filed: %s", err) } else if len(ruleList) > 0 { if err := c.server.UpdateRule(c.Tag, ruleList); err != nil { log.Print(err) } if len(protocolRule) > 0 { if err := c.server.UpdateProtocolRule(c.Tag, protocolRule); err != nil { log.Print(err) } } } } c.nodeInfoMonitorPeriodic = &task.Periodic{ Interval: time.Duration(c.config.UpdatePeriodic) * time.Second, Execute: c.nodeInfoMonitor, } c.userReportPeriodic = &task.Periodic{ Interval: time.Duration(c.config.UpdatePeriodic) * time.Second, Execute: c.userInfoMonitor, } log.Printf("[%s: %d] Start monitor node status", c.nodeInfo.NodeType, c.nodeInfo.NodeId) // delay to start nodeInfoMonitor go func() { time.Sleep(time.Duration(c.config.UpdatePeriodic) * time.Second) _ = c.nodeInfoMonitorPeriodic.Start() }() log.Printf("[%s: %d] Start report node status", c.nodeInfo.NodeType, c.nodeInfo.NodeId) // delay to start userReport go func() { time.Sleep(time.Duration(c.config.UpdatePeriodic) * time.Second) _ = c.userReportPeriodic.Start() }() if c.config.EnableIpRecorder { c.onlineIpReportPeriodic = &task.Periodic{ Interval: time.Duration(c.config.IpRecorderConfig.Periodic) * time.Second, Execute: c.onlineIpReport, } go func() { time.Sleep(time.Duration(c.config.UpdatePeriodic) * time.Second) _ = c.onlineIpReportPeriodic.Start() }() log.Printf("[%s: %d] Start report online ip", c.nodeInfo.NodeType, c.nodeInfo.NodeId) } // delay to start onlineIpReport runtime.GC() return nil } // Close implement the Close() function of the service interface func (c *Node) Close() error { if c.nodeInfoMonitorPeriodic != nil { err := c.nodeInfoMonitorPeriodic.Close() if err != nil { log.Panicf("node info periodic close failed: %s", err) } } if c.nodeInfoMonitorPeriodic != nil { err := c.userReportPeriodic.Close() if err != nil { log.Panicf("user report periodic close failed: %s", err) } } if c.onlineIpReportPeriodic != nil { err := c.onlineIpReportPeriodic.Close() if err != nil { log.Panicf("online ip report periodic close failed: %s", err) } } return nil } func (c *Node) nodeInfoMonitor() (err error) { // First fetch Node Info newNodeInfo, err := c.apiClient.GetNodeInfo() if err != nil { log.Print(err) return nil } var nodeInfoChanged = false // If nodeInfo changed if newNodeInfo != nil { if c.nodeInfo.SS == nil || !reflect.DeepEqual(c.nodeInfo.SS, newNodeInfo.SS) { // Remove old tag oldtag := c.Tag err := c.removeOldTag(oldtag) if err != nil { log.Print(err) return nil } // Add new tag c.nodeInfo = newNodeInfo c.Tag = c.buildNodeTag() err = c.addNewTag(newNodeInfo) if err != nil { log.Print(err) return nil } nodeInfoChanged = true // Remove Old limiter if err = c.server.DeleteInboundLimiter(oldtag); err != nil { log.Print(err) return nil } } } // Check Rule if !c.config.DisableGetRule { if ruleList, protocolRule, err := c.apiClient.GetNodeRule(); err != nil { log.Printf("Get rule list filed: %s", err) } else if len(ruleList) > 0 { if err := c.server.UpdateRule(c.Tag, ruleList); err != nil { log.Print(err) } if len(protocolRule) > 0 { if err := c.server.UpdateProtocolRule(c.Tag, protocolRule); err != nil { log.Print(err) } } } } // Check Cert if c.nodeInfo.EnableTls && c.config.CertConfig.CertMode != "none" && (c.config.CertConfig.CertMode == "dns" || c.config.CertConfig.CertMode == "http") { lego, err := legoCmd.New() if err != nil { log.Print(err) } // Core-core supports the OcspStapling certification hot renew _, _, err = lego.RenewCert(c.config.CertConfig.CertDomain, c.config.CertConfig.Email, c.config.CertConfig.CertMode, c.config.CertConfig.Provider, c.config.CertConfig.DNSEnv) if err != nil { log.Print(err) } } // Update User newUserInfo, err := c.apiClient.GetUserList() if err != nil { log.Print(err) return nil } if nodeInfoChanged { c.userList = newUserInfo newUserInfo = nil err = c.addNewUser(c.userList, newNodeInfo) if err != nil { log.Print(err) return nil } newNodeInfo = nil // Add Limiter if err := c.server.AddInboundLimiter(c.Tag, c.nodeInfo, c.userList); err != nil { log.Print(err) return nil } runtime.GC() } else { deleted, added := compareUserList(c.userList, newUserInfo) if len(deleted) > 0 { deletedEmail := make([]string, len(deleted)) for i := range deleted { deletedEmail[i] = fmt.Sprintf("%s|%s|%d", c.Tag, (deleted)[i].GetUserEmail(), (deleted)[i].UID) } err := c.server.RemoveUsers(deletedEmail, c.Tag) if err != nil { log.Print(err) } } if len(added) > 0 { err = c.addNewUser(added, c.nodeInfo) if err != nil { log.Print(err) } // Update Limiter if err := c.server.UpdateInboundLimiter(c.Tag, c.nodeInfo, added); err != nil { log.Print(err) } } log.Printf("[%s: %d] %d user deleted, %d user added", c.nodeInfo.NodeType, c.nodeInfo.NodeId, len(deleted), len(added)) c.userList = newUserInfo newUserInfo = nil runtime.GC() } return nil } func (c *Node) removeOldTag(oldtag string) (err error) { err = c.server.RemoveInbound(oldtag) if err != nil { return err } err = c.server.RemoveOutbound(oldtag) if err != nil { return err } return nil } func (c *Node) addNewTag(newNodeInfo *panel.NodeInfo) (err error) { inboundConfig, err := InboundBuilder(c.config, newNodeInfo, c.Tag) if err != nil { return err } err = c.server.AddInbound(inboundConfig) if err != nil { return err } outBoundConfig, err := OutboundBuilder(c.config, newNodeInfo, c.Tag) if err != nil { return err } err = c.server.AddOutbound(outBoundConfig) if err != nil { return err } return nil } func (c *Node) addNewUser(userInfo []panel.UserInfo, nodeInfo *panel.NodeInfo) (err error) { users := make([]*protocol.User, 0) if nodeInfo.NodeType == "V2ray" { if nodeInfo.EnableVless { users = c.buildVlessUsers(userInfo) } else { alterID := 0 alterID = (userInfo)[0].V2rayUser.AlterId if alterID >= 0 && alterID < math.MaxUint16 { users = c.buildVmessUsers(userInfo, uint16(alterID)) } else { users = c.buildVmessUsers(userInfo, 0) return fmt.Errorf("AlterID should between 0 to 1<<16 - 1, set it to 0 for now") } } } else if nodeInfo.NodeType == "Trojan" { users = c.buildTrojanUsers(userInfo) } else if nodeInfo.NodeType == "Shadowsocks" { users = c.buildSSUsers(userInfo, getCipherFromString(nodeInfo.SS.CypherMethod)) } else { return fmt.Errorf("unsupported node type: %s", nodeInfo.NodeType) } err = c.server.AddUsers(users, c.Tag) if err != nil { return err } log.Printf("[%s: %d] Added %d new users", c.nodeInfo.NodeType, c.nodeInfo.NodeId, len(userInfo)) return nil } func compareUserList(old, new []panel.UserInfo) (deleted, added []panel.UserInfo) { tmp := map[string]struct{}{} tmp2 := map[string]struct{}{} for i := range old { tmp[(old)[i].GetUserEmail()] = struct{}{} } l := len(tmp) for i := range new { e := (new)[i].GetUserEmail() tmp[e] = struct{}{} tmp2[e] = struct{}{} if l != len(tmp) { added = append(added, (new)[i]) l++ } } tmp = nil l = len(tmp2) for i := range old { tmp2[(old)[i].GetUserEmail()] = struct{}{} if l != len(tmp2) { deleted = append(deleted, (old)[i]) l++ } } return deleted, added } func (c *Node) userInfoMonitor() (err error) { // Get User traffic userTraffic := make([]panel.UserTraffic, 0) for i := range c.userList { up, down := c.server.GetUserTraffic(c.buildUserTag(&(c.userList)[i])) if up > 0 || down > 0 { userTraffic = append(userTraffic, panel.UserTraffic{ UID: (c.userList)[i].UID, Upload: up, Download: down}) } } if len(userTraffic) > 0 && !c.config.DisableUploadTraffic { err = c.apiClient.ReportUserTraffic(userTraffic) if err != nil { log.Print(err) } else { log.Printf("[%s: %d] Report %d online users", c.nodeInfo.NodeType, c.nodeInfo.NodeId, len(userTraffic)) } } userTraffic = nil if !c.config.EnableIpRecorder { c.server.ClearOnlineIp(c.Tag) } runtime.GC() return nil } func (c *Node) onlineIpReport() (err error) { onlineIp, err := c.server.ListOnlineIp(c.Tag) if err != nil { log.Print(err) return nil } rsp, err := resty.New().SetTimeout(time.Duration(c.config.IpRecorderConfig.Timeout) * time.Second). R(). SetBody(onlineIp). Post(c.config.IpRecorderConfig.Url + "/api/v1/SyncOnlineIp?token=" + c.config.IpRecorderConfig.Token) if err != nil { log.Print(err) c.server.ClearOnlineIp(c.Tag) return nil } log.Printf("[Node: %d] Report %d online ip", c.nodeInfo.NodeId, len(onlineIp)) if rsp.StatusCode() == 200 { onlineIp = []dispatcher.UserIp{} err := json.Unmarshal(rsp.Body(), &onlineIp) if err != nil { log.Print(err) c.server.ClearOnlineIp(c.Tag) return nil } c.server.UpdateOnlineIp(c.Tag, onlineIp) log.Printf("[Node: %d] Updated %d online ip", c.nodeInfo.NodeId, len(onlineIp)) } else { c.server.ClearOnlineIp(c.Tag) } return nil } func (c *Node) buildNodeTag() string { return fmt.Sprintf("%s_%s_%d", c.nodeInfo.NodeType, c.config.ListenIP, c.nodeInfo.NodeId) }