node.go 26 KB

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