1
0

node.go 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928
  1. package service
  2. import (
  3. "context"
  4. "crypto/sha256"
  5. "crypto/tls"
  6. "encoding/base64"
  7. "encoding/json"
  8. "errors"
  9. "fmt"
  10. "net"
  11. "net/http"
  12. "net/url"
  13. "slices"
  14. "strconv"
  15. "strings"
  16. "time"
  17. "github.com/mhsanaei/3x-ui/v3/internal/database"
  18. "github.com/mhsanaei/3x-ui/v3/internal/database/model"
  19. "github.com/mhsanaei/3x-ui/v3/internal/logger"
  20. "github.com/mhsanaei/3x-ui/v3/internal/util/common"
  21. "github.com/mhsanaei/3x-ui/v3/internal/util/json_util"
  22. "github.com/mhsanaei/3x-ui/v3/internal/util/netsafe"
  23. "github.com/mhsanaei/3x-ui/v3/internal/web/runtime"
  24. "github.com/mhsanaei/3x-ui/v3/internal/xray"
  25. "gorm.io/gorm"
  26. )
  27. type HeartbeatPatch struct {
  28. Status string
  29. LastHeartbeat int64
  30. LatencyMs int
  31. XrayVersion string
  32. PanelVersion string
  33. Guid string
  34. CpuPct float64
  35. MemPct float64
  36. UptimeSecs uint64
  37. // NetUp/NetDown are the node's current interface throughput (bytes/sec),
  38. // summed over non-virtual interfaces, read from its status response.
  39. NetUp uint64
  40. NetDown uint64
  41. LastError string
  42. // XrayState and XrayError come from the remote /panel/api/server/status when the
  43. // panel API is reachable. They allow distinguishing panel connectivity from
  44. // Xray core health on the node.
  45. XrayState string
  46. XrayError string
  47. }
  48. type NodeService struct{}
  49. // FetchCertFingerprint connects to the node over HTTPS without verifying the
  50. // certificate and returns the leaf certificate's SHA-256 as base64, so the UI
  51. // can offer a "fetch and pin current certificate" action.
  52. func (s *NodeService) FetchCertFingerprint(ctx context.Context, n *model.Node) (string, error) {
  53. addr, err := netsafe.NormalizeHost(n.Address)
  54. if err != nil {
  55. return "", err
  56. }
  57. scheme := n.Scheme
  58. if scheme != "http" && scheme != "https" {
  59. scheme = "https"
  60. }
  61. if scheme != "https" {
  62. return "", common.NewError("certificate pinning is only available for https nodes")
  63. }
  64. if n.Port <= 0 || n.Port > 65535 {
  65. return "", common.NewError("node port must be 1-65535")
  66. }
  67. probeURL := &url.URL{
  68. Scheme: scheme,
  69. Host: net.JoinHostPort(addr, strconv.Itoa(n.Port)),
  70. Path: normalizeBasePath(n.BasePath) + "panel/api/server/status",
  71. }
  72. req, err := http.NewRequestWithContext(
  73. netsafe.ContextWithAllowPrivate(ctx, n.AllowPrivateAddress),
  74. http.MethodGet, probeURL.String(), nil)
  75. if err != nil {
  76. return "", err
  77. }
  78. client := &http.Client{
  79. Transport: &http.Transport{
  80. DialContext: netsafe.SSRFGuardedDialContext,
  81. TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, // lgtm[go/disabled-certificate-check]
  82. },
  83. }
  84. resp, err := client.Do(req)
  85. if err != nil {
  86. return "", err
  87. }
  88. defer resp.Body.Close()
  89. if resp.TLS == nil || len(resp.TLS.PeerCertificates) == 0 {
  90. return "", common.NewError("node did not present a TLS certificate")
  91. }
  92. sum := sha256.Sum256(resp.TLS.PeerCertificates[0].Raw)
  93. return base64.StdEncoding.EncodeToString(sum[:]), nil
  94. }
  95. func (s *NodeService) GetAll() ([]*model.Node, error) {
  96. db := database.GetDB()
  97. var nodes []*model.Node
  98. err := db.Model(model.Node{}).Order("id asc").Find(&nodes).Error
  99. if err != nil || len(nodes) == 0 {
  100. return nodes, err
  101. }
  102. type inboundRow struct {
  103. Id int
  104. NodeID int `gorm:"column:node_id"`
  105. }
  106. var inboundRows []inboundRow
  107. if err := db.Table("inbounds").
  108. Select("id, node_id").
  109. Where("node_id IS NOT NULL").
  110. Scan(&inboundRows).Error; err != nil {
  111. return nodes, nil
  112. }
  113. if len(inboundRows) == 0 {
  114. return nodes, nil
  115. }
  116. inboundsByNode := make(map[int][]int, len(nodes))
  117. nodeByInbound := make(map[int]int, len(inboundRows))
  118. for _, row := range inboundRows {
  119. inboundsByNode[row.NodeID] = append(inboundsByNode[row.NodeID], row.Id)
  120. nodeByInbound[row.Id] = row.NodeID
  121. }
  122. type clientCountRow struct {
  123. NodeID int `gorm:"column:node_id"`
  124. Count int `gorm:"column:count"`
  125. }
  126. var clientCounts []clientCountRow
  127. if err := db.Raw(`
  128. SELECT inbounds.node_id AS node_id, COUNT(DISTINCT client_inbounds.client_id) AS count
  129. FROM inbounds
  130. JOIN client_inbounds ON client_inbounds.inbound_id = inbounds.id
  131. WHERE inbounds.node_id IS NOT NULL
  132. GROUP BY inbounds.node_id
  133. `).Scan(&clientCounts).Error; err == nil {
  134. for _, row := range clientCounts {
  135. for _, n := range nodes {
  136. if n.Id == row.NodeID {
  137. n.ClientCount = row.Count
  138. break
  139. }
  140. }
  141. }
  142. }
  143. now := time.Now().UnixMilli()
  144. type trafficRow struct {
  145. InboundID int `gorm:"column:inbound_id"`
  146. Email string
  147. Enable bool
  148. Total int64
  149. Up int64
  150. Down int64
  151. ExpiryTime int64 `gorm:"column:expiry_time"`
  152. }
  153. var trafficRows []trafficRow
  154. inboundIDs := make([]int, 0, len(nodeByInbound))
  155. for id := range nodeByInbound {
  156. inboundIDs = append(inboundIDs, id)
  157. }
  158. // Chunk the IN clause to avoid "too many SQL variables" on SQLite
  159. // when there are many node-owned inbounds (common with many nodes).
  160. // sqliteMaxVars is defined in this package (inbound.go).
  161. for _, batch := range chunkInts(inboundIDs, sqliteMaxVars) {
  162. var page []trafficRow
  163. if err := db.Table("client_traffics").
  164. Select("inbound_id, email, enable, total, up, down, expiry_time").
  165. Where("inbound_id IN ?", batch).
  166. Scan(&page).Error; err == nil {
  167. trafficRows = append(trafficRows, page...)
  168. }
  169. }
  170. depletedByNode := make(map[int]int)
  171. if len(trafficRows) > 0 {
  172. for _, row := range trafficRows {
  173. nodeID, ok := nodeByInbound[row.InboundID]
  174. if !ok {
  175. continue
  176. }
  177. expired := row.ExpiryTime > 0 && row.ExpiryTime <= now
  178. exhausted := row.Total > 0 && row.Up+row.Down >= row.Total
  179. if expired || exhausted || !row.Enable {
  180. depletedByNode[nodeID]++
  181. }
  182. }
  183. }
  184. onlineByGuid := s.onlineEmailsByGuid()
  185. for _, n := range nodes {
  186. n.InboundCount = len(inboundsByNode[n.Id])
  187. n.DepletedCount = depletedByNode[n.Id]
  188. // Online is attributed to the node that physically hosts the client
  189. // (by GUID): a client on a sub-node counts under the sub-node, not
  190. // the intermediate node it syncs through (#4983).
  191. n.OnlineCount = len(onlineByGuid[effectiveNodeGuid(n)])
  192. }
  193. return nodes, nil
  194. }
  195. func (s *NodeService) onlineEmailsByGuid() map[string]map[string]struct{} {
  196. svc := InboundService{}
  197. byGuid := svc.GetOnlineClientsByGuid()
  198. out := make(map[string]map[string]struct{}, len(byGuid))
  199. for guid, emails := range byGuid {
  200. set := make(map[string]struct{}, len(emails))
  201. for _, email := range emails {
  202. set[email] = struct{}{}
  203. }
  204. out[guid] = set
  205. }
  206. return out
  207. }
  208. // effectiveNodeGuid is a node's stable online-attribution key: its reported
  209. // panelGuid, or a master-local synthetic id when the node is an old build that
  210. // hasn't reported one yet (#4983).
  211. func effectiveNodeGuid(n *model.Node) string {
  212. if n.Guid != "" {
  213. return n.Guid
  214. }
  215. return synthNodeGuid(n.Id)
  216. }
  217. func (s *NodeService) GetById(id int) (*model.Node, error) {
  218. db := database.GetDB()
  219. n := &model.Node{}
  220. if err := db.Model(model.Node{}).Where("id = ?", id).First(n).Error; err != nil {
  221. return nil, err
  222. }
  223. return n, nil
  224. }
  225. // NodeExists reports whether a node with the given id exists on this panel.
  226. // Used to drop stale, cross-panel node references on inbound import. A Count
  227. // query distinguishes "no such node" (count 0, no error) from a real DB error.
  228. func (s *NodeService) NodeExists(id int) (bool, error) {
  229. if id <= 0 {
  230. return false, nil
  231. }
  232. var count int64
  233. if err := database.GetDB().Model(model.Node{}).Where("id = ?", id).Count(&count).Error; err != nil {
  234. return false, err
  235. }
  236. return count > 0, nil
  237. }
  238. func normalizeBasePath(p string) string {
  239. p = strings.TrimSpace(p)
  240. if p == "" {
  241. return "/"
  242. }
  243. if !strings.HasPrefix(p, "/") {
  244. p = "/" + p
  245. }
  246. if !strings.HasSuffix(p, "/") {
  247. p = p + "/"
  248. }
  249. return p
  250. }
  251. func (s *NodeService) normalize(n *model.Node) error {
  252. n.Name = strings.TrimSpace(n.Name)
  253. n.ApiToken = strings.TrimSpace(n.ApiToken)
  254. if n.Name == "" {
  255. return common.NewError("node name is required")
  256. }
  257. addr, err := netsafe.NormalizeHost(n.Address)
  258. if err != nil {
  259. return common.NewError(err.Error())
  260. }
  261. n.Address = addr
  262. if n.Port <= 0 || n.Port > 65535 {
  263. return common.NewError("node port must be 1-65535")
  264. }
  265. if n.Scheme != "http" && n.Scheme != "https" {
  266. n.Scheme = "https"
  267. }
  268. if n.TlsVerifyMode != "skip" && n.TlsVerifyMode != "pin" && n.TlsVerifyMode != "mtls" {
  269. n.TlsVerifyMode = "verify"
  270. }
  271. if n.TlsVerifyMode == "mtls" && n.Scheme != "https" {
  272. return common.NewError("mtls requires the node scheme to be https")
  273. }
  274. n.PinnedCertSha256 = strings.TrimSpace(n.PinnedCertSha256)
  275. if n.InboundSyncMode != "selected" {
  276. n.InboundSyncMode = "all"
  277. n.InboundTags = nil
  278. } else {
  279. seen := make(map[string]struct{}, len(n.InboundTags))
  280. tags := make([]string, 0, len(n.InboundTags))
  281. for _, tag := range n.InboundTags {
  282. tag = strings.TrimSpace(tag)
  283. if tag == "" {
  284. continue
  285. }
  286. if _, ok := seen[tag]; ok {
  287. continue
  288. }
  289. seen[tag] = struct{}{}
  290. tags = append(tags, tag)
  291. }
  292. n.InboundTags = tags
  293. }
  294. if n.TlsVerifyMode == "pin" {
  295. if _, err := runtime.DecodeCertPin(n.PinnedCertSha256); err != nil {
  296. return common.NewError(err.Error())
  297. }
  298. }
  299. n.BasePath = normalizeBasePath(n.BasePath)
  300. return nil
  301. }
  302. func (s *NodeService) Create(n *model.Node) error {
  303. if err := s.normalize(n); err != nil {
  304. return err
  305. }
  306. db := database.GetDB()
  307. return db.Create(n).Error
  308. }
  309. func (s *NodeService) Update(id int, in *model.Node) error {
  310. if err := s.normalize(in); err != nil {
  311. return err
  312. }
  313. inboundTagsJSON, err := json.Marshal(in.InboundTags)
  314. if err != nil {
  315. return err
  316. }
  317. db := database.GetDB()
  318. existing := &model.Node{}
  319. if err := db.Where("id = ?", id).First(existing).Error; err != nil {
  320. return err
  321. }
  322. updates := map[string]any{
  323. "name": in.Name,
  324. "remark": in.Remark,
  325. "scheme": in.Scheme,
  326. "address": in.Address,
  327. "port": in.Port,
  328. "base_path": in.BasePath,
  329. "api_token": in.ApiToken,
  330. "enable": in.Enable,
  331. "allow_private_address": in.AllowPrivateAddress,
  332. "tls_verify_mode": in.TlsVerifyMode,
  333. "pinned_cert_sha256": in.PinnedCertSha256,
  334. "inbound_sync_mode": in.InboundSyncMode,
  335. "inbound_tags": string(inboundTagsJSON),
  336. "outbound_tag": in.OutboundTag,
  337. }
  338. if err := db.Model(model.Node{}).Where("id = ?", id).Updates(updates).Error; err != nil {
  339. return err
  340. }
  341. if dErr := s.MarkNodeDirty(id); dErr != nil {
  342. logger.Warning("mark node dirty after update failed:", dErr)
  343. }
  344. if mgr := runtime.GetManager(); mgr != nil {
  345. mgr.InvalidateNode(id)
  346. }
  347. return nil
  348. }
  349. func (s *NodeService) GetRemoteInboundOptions(ctx context.Context, n *model.Node) ([]runtime.RemoteInboundOption, error) {
  350. if err := s.normalize(n); err != nil {
  351. return nil, err
  352. }
  353. if n.OutboundTag == "" {
  354. return runtime.NewRemote(n, nil).ListInboundOptions(ctx)
  355. }
  356. // Mirror ProbeWithOutbound: a node being added/edited has no persistent
  357. // egress bridge yet, so route the list call through a temporary one or the
  358. // remote panel stays unreachable and the request times out.
  359. var options []runtime.RemoteInboundOption
  360. var err error
  361. s.withOutboundBridge(n.Id, n.OutboundTag, func(proxyURL string) {
  362. options, err = runtime.NewRemote(n, staticEgressResolver(proxyURL)).ListInboundOptions(ctx)
  363. })
  364. return options, err
  365. }
  366. // staticEgressResolver hands a fixed proxy URL to runtime.NewRemote. An empty
  367. // string yields a direct connection, so it doubles as the graceful fallback
  368. // when a temporary bridge can't be built.
  369. type staticEgressResolver string
  370. func (r staticEgressResolver) NodeEgressProxyURL(int) string { return string(r) }
  371. // EnsureInboundTagAllowed adds a panel-managed inbound's tag to the node's
  372. // selection when the node syncs in "selected" mode. Without it, the next
  373. // traffic sync would filter the tag out of the snapshot and the orphan sweep
  374. // would silently delete the central row the panel just created or renamed.
  375. // Tags are only ever added (never removed): on a rename the node may keep
  376. // reporting the old tag until the remote update lands, and a leftover entry
  377. // that matches nothing is harmless.
  378. func (s *NodeService) EnsureInboundTagAllowed(nodeID int, tag string) error {
  379. tag = strings.TrimSpace(tag)
  380. if nodeID <= 0 || tag == "" {
  381. return nil
  382. }
  383. db := database.GetDB()
  384. node := &model.Node{}
  385. if err := db.Where("id = ?", nodeID).First(node).Error; err != nil {
  386. return err
  387. }
  388. if node.InboundSyncMode != "selected" {
  389. return nil
  390. }
  391. if slices.Contains(node.InboundTags, tag) {
  392. return nil
  393. }
  394. buf, err := json.Marshal(append(node.InboundTags, tag))
  395. if err != nil {
  396. return err
  397. }
  398. return db.Model(model.Node{}).Where("id = ?", nodeID).
  399. Updates(map[string]any{"inbound_tags": string(buf)}).Error
  400. }
  401. func FilterNodeSnapshot(n *model.Node, snap *runtime.TrafficSnapshot) {
  402. if n == nil || snap == nil || n.InboundSyncMode != "selected" {
  403. return
  404. }
  405. allowed := make(map[string]struct{}, len(n.InboundTags))
  406. for _, tag := range n.InboundTags {
  407. allowed[tag] = struct{}{}
  408. }
  409. filtered := make([]*model.Inbound, 0, len(snap.Inbounds))
  410. for _, inbound := range snap.Inbounds {
  411. if inbound == nil {
  412. continue
  413. }
  414. if _, ok := allowed[inbound.Tag]; ok {
  415. filtered = append(filtered, inbound)
  416. }
  417. }
  418. snap.Inbounds = filtered
  419. }
  420. func (s *NodeService) Delete(id int) error {
  421. db := database.GetDB()
  422. // Refuse to delete a node that still owns inbounds: dropping the node row
  423. // while inbounds keep its node_id leaves orphaned, dangling references that
  424. // confuse node sync, subscriptions and cleanup. The operator must detach or
  425. // remove those inbounds first. (DB-002)
  426. var attached int64
  427. if err := db.Model(&model.Inbound{}).Where("node_id = ?", id).Count(&attached).Error; err != nil {
  428. return err
  429. }
  430. if attached > 0 {
  431. return common.NewError(fmt.Sprintf("cannot delete node: %d inbound(s) still attached to it; detach or delete them first", attached))
  432. }
  433. // Capture the node's guid before deleting the row so we can drop its per-node
  434. // IP attribution (NodeClientIp is keyed by guid, not node id).
  435. var guid string
  436. var n model.Node
  437. if err := db.Select("guid").Where("id = ?", id).First(&n).Error; err == nil {
  438. guid = n.Guid
  439. }
  440. // Delete the node row and its per-node child rows atomically. Remove the
  441. // children (traffic baselines, IP attribution) before the parent node row so
  442. // the ordering already matches a future ON DELETE constraint. Delete stays
  443. // tolerant of a missing node row so it can still clean up orphaned baselines.
  444. if err := db.Transaction(func(tx *gorm.DB) error {
  445. if err := tx.Where("node_id = ?", id).Delete(&model.NodeClientTraffic{}).Error; err != nil {
  446. return err
  447. }
  448. if guid != "" {
  449. if err := tx.Where("node_guid = ?", guid).Delete(&model.NodeClientIp{}).Error; err != nil {
  450. return err
  451. }
  452. }
  453. return tx.Where("id = ?", id).Delete(&model.Node{}).Error
  454. }); err != nil {
  455. return err
  456. }
  457. if mgr := runtime.GetManager(); mgr != nil {
  458. mgr.InvalidateNode(id)
  459. }
  460. nodeMetrics.drop(nodeMetricKey(id, "cpu"))
  461. nodeMetrics.drop(nodeMetricKey(id, "mem"))
  462. return nil
  463. }
  464. func (s *NodeService) SetEnable(id int, enable bool) error {
  465. db := database.GetDB()
  466. if err := db.Model(model.Node{}).Where("id = ?", id).Update("enable", enable).Error; err != nil {
  467. return err
  468. }
  469. if mgr := runtime.GetManager(); mgr != nil {
  470. mgr.InvalidateNode(id)
  471. }
  472. return nil
  473. }
  474. // GetWebCertFiles asks a node for its own web TLS certificate/key file paths,
  475. // used by "Set Cert from Panel" so a node-assigned inbound gets paths that
  476. // exist on the node rather than the central panel. See issue #4854.
  477. func (s *NodeService) GetWebCertFiles(id int) (*runtime.WebCertFiles, error) {
  478. n, err := s.GetById(id)
  479. if err != nil || n == nil {
  480. return nil, fmt.Errorf("node not found")
  481. }
  482. if !n.Enable {
  483. return nil, fmt.Errorf("node is disabled")
  484. }
  485. mgr := runtime.GetManager()
  486. if mgr == nil {
  487. return nil, fmt.Errorf("runtime manager unavailable")
  488. }
  489. remote, err := mgr.RemoteFor(n)
  490. if err != nil {
  491. return nil, err
  492. }
  493. ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
  494. defer cancel()
  495. return remote.GetWebCertFiles(ctx)
  496. }
  497. // NodeUpdateResult reports the outcome of triggering a panel self-update on one
  498. // node so the UI can show per-node success/failure for a bulk request.
  499. type NodeUpdateResult struct {
  500. Id int `json:"id"`
  501. Name string `json:"name"`
  502. OK bool `json:"ok"`
  503. Error string `json:"error,omitempty"`
  504. }
  505. // UpdatePanels triggers the official self-updater on each given node. Only
  506. // enabled, online nodes are eligible — an offline node can't be reached, so it
  507. // is reported as skipped rather than silently dropped.
  508. func (s *NodeService) UpdatePanels(ids []int) ([]NodeUpdateResult, error) {
  509. mgr := runtime.GetManager()
  510. if mgr == nil {
  511. return nil, fmt.Errorf("runtime manager unavailable")
  512. }
  513. results := make([]NodeUpdateResult, 0, len(ids))
  514. for _, id := range ids {
  515. n, err := s.GetById(id)
  516. if err != nil || n == nil {
  517. results = append(results, NodeUpdateResult{Id: id, OK: false, Error: "node not found"})
  518. continue
  519. }
  520. res := NodeUpdateResult{Id: id, Name: n.Name}
  521. switch {
  522. case !n.Enable:
  523. res.Error = "node is disabled"
  524. case n.Status != "online":
  525. res.Error = "node is offline"
  526. default:
  527. remote, remoteErr := mgr.RemoteFor(n)
  528. if remoteErr != nil {
  529. res.Error = remoteErr.Error()
  530. break
  531. }
  532. ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
  533. updErr := remote.UpdatePanel(ctx)
  534. cancel()
  535. if updErr != nil {
  536. res.Error = updErr.Error()
  537. } else {
  538. res.OK = true
  539. }
  540. }
  541. results = append(results, res)
  542. }
  543. return results, nil
  544. }
  545. func (s *NodeService) UpdateHeartbeat(id int, p HeartbeatPatch) error {
  546. db := database.GetDB()
  547. updates := map[string]any{
  548. "status": p.Status,
  549. "last_heartbeat": p.LastHeartbeat,
  550. "latency_ms": p.LatencyMs,
  551. "xray_version": p.XrayVersion,
  552. "panel_version": p.PanelVersion,
  553. "cpu_pct": p.CpuPct,
  554. "mem_pct": p.MemPct,
  555. "uptime_secs": p.UptimeSecs,
  556. "net_up": p.NetUp,
  557. "net_down": p.NetDown,
  558. "last_error": p.LastError,
  559. "xray_state": p.XrayState,
  560. "xray_error": p.XrayError,
  561. }
  562. // Only learn the GUID; never clear a known one if an old-build node (or a
  563. // failed probe) reports none, so the stable identity survives blips.
  564. if p.Guid != "" {
  565. updates["guid"] = p.Guid
  566. }
  567. if err := db.Model(model.Node{}).Where("id = ?", id).Updates(updates).Error; err != nil {
  568. return err
  569. }
  570. if p.Status == "online" {
  571. now := time.Unix(p.LastHeartbeat, 0)
  572. nodeMetrics.append(nodeMetricKey(id, "cpu"), now, p.CpuPct)
  573. nodeMetrics.append(nodeMetricKey(id, "mem"), now, p.MemPct)
  574. nodeMetrics.append(nodeMetricKey(id, "netUp"), now, float64(p.NetUp))
  575. nodeMetrics.append(nodeMetricKey(id, "netDown"), now, float64(p.NetDown))
  576. }
  577. return nil
  578. }
  579. func (s *NodeService) MarkNodeDirty(id int) error {
  580. if id <= 0 {
  581. return nil
  582. }
  583. return database.GetDB().Model(model.Node{}).
  584. Where("id = ?", id).
  585. Updates(map[string]any{
  586. "config_dirty": true,
  587. "config_dirty_at": time.Now().UnixMilli(),
  588. }).Error
  589. }
  590. func (s *NodeService) ClearNodeDirty(id int, dirtyAt int64) error {
  591. if id <= 0 {
  592. return nil
  593. }
  594. return database.GetDB().Model(model.Node{}).
  595. Where("id = ? AND config_dirty_at = ?", id, dirtyAt).
  596. Update("config_dirty", false).Error
  597. }
  598. func (s *NodeService) NodeSyncState(id int) (enabled bool, status string, dirty bool, dirtyAt int64, err error) {
  599. if id <= 0 {
  600. return false, "", false, 0, errors.New("invalid node id")
  601. }
  602. var row model.Node
  603. err = database.GetDB().Model(model.Node{}).
  604. Select("enable", "status", "config_dirty", "config_dirty_at").
  605. Where("id = ?", id).
  606. First(&row).Error
  607. if err != nil {
  608. return false, "", false, 0, err
  609. }
  610. return row.Enable, row.Status, row.ConfigDirty, row.ConfigDirtyAt, nil
  611. }
  612. func (s *NodeService) IsNodePending(id int) bool {
  613. enabled, status, dirty, _, err := s.NodeSyncState(id)
  614. if err != nil {
  615. return false
  616. }
  617. return !enabled || status != "online" || dirty
  618. }
  619. func nodeMetricKey(id int, metric string) string {
  620. return "node:" + strconv.Itoa(id) + ":" + metric
  621. }
  622. func (s *NodeService) AggregateNodeMetric(id int, metric string, bucketSeconds int, maxPoints int) []map[string]any {
  623. return nodeMetrics.aggregate(nodeMetricKey(id, metric), bucketSeconds, maxPoints)
  624. }
  625. func (s *NodeService) Probe(ctx context.Context, n *model.Node) (HeartbeatPatch, error) {
  626. proxyURL := ""
  627. if n.OutboundTag != "" {
  628. if mgr := runtime.GetManager(); mgr != nil {
  629. proxyURL = mgr.NodeEgressProxyURL(n.Id)
  630. }
  631. }
  632. return s.probe(ctx, n, proxyURL)
  633. }
  634. func (s *NodeService) ProbeWithOutbound(ctx context.Context, n *model.Node, outboundTag string) (HeartbeatPatch, error) {
  635. if outboundTag == "" {
  636. return s.Probe(ctx, n)
  637. }
  638. var patch HeartbeatPatch
  639. var err error
  640. s.withOutboundBridge(n.Id, outboundTag, func(proxyURL string) {
  641. if proxyURL == "" {
  642. patch, err = s.Probe(ctx, n)
  643. return
  644. }
  645. patch, err = s.probe(ctx, n, proxyURL)
  646. })
  647. return patch, err
  648. }
  649. // withOutboundBridge stands up a temporary loopback SOCKS5 inbound in the
  650. // running Xray, routes it through outboundTag, and runs fn with the bridge's
  651. // proxy URL before tearing it down. It is used to reach a node through its
  652. // connection outbound before the persistent egress bridge has been injected
  653. // into the config (e.g. while the node is still being added or edited). When
  654. // Xray isn't running or the bridge can't be built, fn runs with an empty
  655. // proxyURL so callers fall back to a direct connection.
  656. func (s *NodeService) withOutboundBridge(nodeID int, outboundTag string, fn func(proxyURL string)) {
  657. proc := XrayProcess()
  658. if proc == nil || !proc.IsRunning() {
  659. fn("")
  660. return
  661. }
  662. apiPort := proc.GetAPIPort()
  663. if apiPort <= 0 {
  664. fn("")
  665. return
  666. }
  667. listener, err := net.Listen("tcp", "127.0.0.1:0")
  668. if err != nil {
  669. fn("")
  670. return
  671. }
  672. port := listener.Addr().(*net.TCPAddr).Port
  673. listener.Close()
  674. tag := fmt.Sprintf("node-test-%d-%d", nodeID, time.Now().UnixNano())
  675. proxyURL := fmt.Sprintf("socks5://127.0.0.1:%d", port)
  676. inboundJSON, err := json.Marshal(xray.InboundConfig{
  677. Listen: json_util.RawMessage(`"127.0.0.1"`),
  678. Port: port,
  679. Protocol: "socks",
  680. Settings: json_util.RawMessage(`{"auth":"noauth","udp":false}`),
  681. Tag: tag,
  682. })
  683. if err != nil {
  684. fn("")
  685. return
  686. }
  687. cfg := proc.GetConfig()
  688. routing := map[string]any{}
  689. if len(cfg.RouterConfig) > 0 {
  690. _ = json.Unmarshal(cfg.RouterConfig, &routing)
  691. }
  692. rules, _ := routing["rules"].([]any)
  693. rule := map[string]any{
  694. "type": "field",
  695. "inboundTag": []any{tag},
  696. }
  697. if routingTagIsBalancer(routing, outboundTag) {
  698. rule["balancerTag"] = outboundTag
  699. } else {
  700. rule["outboundTag"] = outboundTag
  701. }
  702. routing["rules"] = append([]any{rule}, rules...)
  703. routingJSON, err := json.Marshal(routing)
  704. if err != nil {
  705. fn("")
  706. return
  707. }
  708. originalRoutingJSON := cfg.RouterConfig
  709. api := xray.XrayAPI{}
  710. if err := api.Init(apiPort); err != nil {
  711. fn("")
  712. return
  713. }
  714. defer api.Close()
  715. if err := api.AddInbound(inboundJSON); err != nil {
  716. fn("")
  717. return
  718. }
  719. defer func() {
  720. if err := api.DelInbound(tag); err != nil {
  721. logger.Warning("remove temp node bridge inbound failed:", err)
  722. }
  723. }()
  724. if err := api.ApplyRoutingConfig(routingJSON); err != nil {
  725. fn("")
  726. return
  727. }
  728. defer func() {
  729. restore := originalRoutingJSON
  730. if len(restore) == 0 {
  731. restore = []byte("{}")
  732. }
  733. if err := api.ApplyRoutingConfig(restore); err != nil {
  734. logger.Warning("restore routing after node bridge failed:", err)
  735. }
  736. }()
  737. fn(proxyURL)
  738. }
  739. func (s *NodeService) probe(ctx context.Context, n *model.Node, proxyURL string) (HeartbeatPatch, error) {
  740. patch := HeartbeatPatch{LastHeartbeat: time.Now().Unix()}
  741. addr, err := netsafe.NormalizeHost(n.Address)
  742. if err != nil {
  743. patch.LastError = err.Error()
  744. return patch, err
  745. }
  746. scheme := n.Scheme
  747. if scheme != "http" && scheme != "https" {
  748. scheme = "https"
  749. }
  750. if n.Port <= 0 || n.Port > 65535 {
  751. patch.LastError = "node port must be 1-65535"
  752. return patch, errors.New(patch.LastError)
  753. }
  754. probeURL := &url.URL{
  755. Scheme: scheme,
  756. Host: net.JoinHostPort(addr, strconv.Itoa(n.Port)),
  757. Path: normalizeBasePath(n.BasePath) + "panel/api/server/status",
  758. }
  759. req, err := http.NewRequestWithContext(
  760. netsafe.ContextWithAllowPrivate(ctx, n.AllowPrivateAddress),
  761. http.MethodGet, probeURL.String(), nil)
  762. if err != nil {
  763. patch.LastError = err.Error()
  764. return patch, err
  765. }
  766. if n.ApiToken != "" {
  767. req.Header.Set("Authorization", "Bearer "+n.ApiToken)
  768. }
  769. req.Header.Set("Accept", "application/json")
  770. client, err := runtime.HTTPClientForNode(n, proxyURL)
  771. if err != nil {
  772. patch.LastError = err.Error()
  773. return patch, err
  774. }
  775. start := time.Now()
  776. resp, err := client.Do(req)
  777. if err != nil {
  778. patch.LastError = err.Error()
  779. return patch, err
  780. }
  781. defer resp.Body.Close()
  782. patch.LatencyMs = int(time.Since(start) / time.Millisecond)
  783. if resp.StatusCode != http.StatusOK {
  784. patch.LastError = fmt.Sprintf("HTTP %d from remote panel", resp.StatusCode)
  785. return patch, errors.New(patch.LastError)
  786. }
  787. var envelope struct {
  788. Success bool `json:"success"`
  789. Msg string `json:"msg"`
  790. Obj *struct {
  791. CpuPct float64 `json:"cpu"`
  792. Mem struct {
  793. Current uint64 `json:"current"`
  794. Total uint64 `json:"total"`
  795. } `json:"mem"`
  796. Xray struct {
  797. Version string `json:"version"`
  798. State string `json:"state"`
  799. ErrorMsg string `json:"errorMsg"`
  800. } `json:"xray"`
  801. PanelVersion string `json:"panelVersion"`
  802. PanelGuid string `json:"panelGuid"`
  803. Uptime uint64 `json:"uptime"`
  804. NetIO struct {
  805. Up uint64 `json:"up"`
  806. Down uint64 `json:"down"`
  807. } `json:"netIO"`
  808. } `json:"obj"`
  809. }
  810. if err := json.NewDecoder(resp.Body).Decode(&envelope); err != nil {
  811. patch.LastError = "decode response: " + err.Error()
  812. return patch, err
  813. }
  814. if !envelope.Success || envelope.Obj == nil {
  815. patch.LastError = "remote returned success=false: " + envelope.Msg
  816. return patch, errors.New(patch.LastError)
  817. }
  818. o := envelope.Obj
  819. patch.CpuPct = o.CpuPct
  820. if o.Mem.Total > 0 {
  821. patch.MemPct = float64(o.Mem.Current) * 100.0 / float64(o.Mem.Total)
  822. }
  823. patch.XrayVersion = o.Xray.Version
  824. patch.XrayState = o.Xray.State
  825. patch.XrayError = o.Xray.ErrorMsg
  826. patch.PanelVersion = o.PanelVersion
  827. patch.Guid = o.PanelGuid
  828. patch.UptimeSecs = o.Uptime
  829. patch.NetUp = o.NetIO.Up
  830. patch.NetDown = o.NetIO.Down
  831. return patch, nil
  832. }
  833. type ProbeResultUI struct {
  834. Status string `json:"status" example:"online"`
  835. LatencyMs int `json:"latencyMs" example:"42"`
  836. XrayVersion string `json:"xrayVersion" example:"25.10.31"`
  837. PanelVersion string `json:"panelVersion" example:"v3.x.x"`
  838. CpuPct float64 `json:"cpuPct" example:"12.5"`
  839. MemPct float64 `json:"memPct" example:"45.2"`
  840. UptimeSecs uint64 `json:"uptimeSecs" example:"86400"`
  841. Error string `json:"error"`
  842. // XrayState/XrayError are populated on successful probes even when the node's
  843. // Xray core is not healthy. The UI uses them for a distinct "panel ok, xray failed" indicator.
  844. XrayState string `json:"xrayState"`
  845. XrayError string `json:"xrayError"`
  846. }
  847. func (p HeartbeatPatch) ToUI(ok bool) ProbeResultUI {
  848. r := ProbeResultUI{
  849. LatencyMs: p.LatencyMs,
  850. XrayVersion: p.XrayVersion,
  851. PanelVersion: p.PanelVersion,
  852. CpuPct: p.CpuPct,
  853. MemPct: p.MemPct,
  854. UptimeSecs: p.UptimeSecs,
  855. Error: FriendlyProbeError(p.LastError),
  856. XrayState: p.XrayState,
  857. XrayError: p.XrayError,
  858. }
  859. if ok {
  860. r.Status = "online"
  861. } else {
  862. r.Status = "offline"
  863. }
  864. return r
  865. }
  866. func FriendlyProbeError(msg string) string {
  867. if strings.Contains(msg, "server gave HTTP response to HTTPS client") {
  868. return "the server speaks HTTP, not HTTPS; set the node scheme to http"
  869. }
  870. return msg
  871. }