node.go 23 KB

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