inbound.go 50 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551
  1. // Package service provides business logic services for the 3x-ui web panel,
  2. // including inbound/outbound management, user administration, settings, and Xray integration.
  3. package service
  4. import (
  5. "context"
  6. "encoding/json"
  7. "errors"
  8. "fmt"
  9. "net"
  10. "sort"
  11. "strings"
  12. "time"
  13. "github.com/mhsanaei/3x-ui/v3/internal/database"
  14. "github.com/mhsanaei/3x-ui/v3/internal/database/model"
  15. "github.com/mhsanaei/3x-ui/v3/internal/logger"
  16. "github.com/mhsanaei/3x-ui/v3/internal/mtproto"
  17. "github.com/mhsanaei/3x-ui/v3/internal/util/common"
  18. "github.com/mhsanaei/3x-ui/v3/internal/util/netsafe"
  19. wgutil "github.com/mhsanaei/3x-ui/v3/internal/util/wireguard"
  20. "github.com/mhsanaei/3x-ui/v3/internal/xray"
  21. "gorm.io/gorm"
  22. "gorm.io/gorm/clause"
  23. )
  24. type InboundService struct {
  25. xrayApi xray.XrayAPI
  26. clientService ClientService
  27. fallbackService FallbackService
  28. }
  29. func normalizeInboundShareAddrStrategy(strategy string) string {
  30. strategy = strings.TrimSpace(strategy)
  31. switch strategy {
  32. case "listen", "custom":
  33. return strategy
  34. default:
  35. return "node"
  36. }
  37. }
  38. func normalizeInboundShareAddress(inbound *model.Inbound) {
  39. if inbound == nil {
  40. return
  41. }
  42. inbound.ShareAddrStrategy = normalizeInboundShareAddrStrategy(inbound.ShareAddrStrategy)
  43. if addr, err := normalizeInboundShareHost(inbound.ShareAddr); err == nil {
  44. inbound.ShareAddr = addr
  45. } else {
  46. inbound.ShareAddr = strings.TrimSpace(inbound.ShareAddr)
  47. }
  48. }
  49. func normalizeInboundShareAddressStrict(inbound *model.Inbound) error {
  50. if inbound == nil {
  51. return nil
  52. }
  53. inbound.ShareAddrStrategy = normalizeInboundShareAddrStrategy(inbound.ShareAddrStrategy)
  54. addr, err := normalizeInboundShareHost(inbound.ShareAddr)
  55. if err != nil {
  56. return common.NewError("shareAddr must be a host or IP without scheme or port")
  57. }
  58. inbound.ShareAddr = addr
  59. return nil
  60. }
  61. func normalizeInboundShareHost(raw string) (string, error) {
  62. addr := strings.TrimSpace(raw)
  63. if addr == "" {
  64. return "", nil
  65. }
  66. if strings.Contains(addr, "://") || strings.HasPrefix(addr, "//") || strings.ContainsAny(addr, "/?#@") {
  67. return "", fmt.Errorf("invalid share address %q", raw)
  68. }
  69. if strings.HasPrefix(addr, "[") {
  70. if !strings.HasSuffix(addr, "]") {
  71. return "", fmt.Errorf("invalid IPv6 host %q", raw)
  72. }
  73. ip := net.ParseIP(addr[1 : len(addr)-1])
  74. if ip == nil || ip.To4() != nil {
  75. return "", fmt.Errorf("invalid IPv6 host %q", raw)
  76. }
  77. return "[" + ip.String() + "]", nil
  78. }
  79. if strings.Contains(addr, ":") {
  80. if _, _, err := net.SplitHostPort(addr); err == nil {
  81. return "", fmt.Errorf("share address must not include port")
  82. }
  83. ip := net.ParseIP(addr)
  84. if ip == nil || ip.To4() != nil {
  85. return "", fmt.Errorf("invalid IPv6 host %q", raw)
  86. }
  87. return "[" + ip.String() + "]", nil
  88. }
  89. host, err := netsafe.NormalizeHost(addr)
  90. if err != nil {
  91. return "", err
  92. }
  93. return host, nil
  94. }
  95. func normalizeInboundShareAddressColumns(tx *gorm.DB) error {
  96. if tx == nil || !tx.Migrator().HasColumn(&model.Inbound{}, "share_addr_strategy") {
  97. return nil
  98. }
  99. strategyExpr := `CASE TRIM(COALESCE(share_addr_strategy, '')) WHEN 'listen' THEN 'listen' WHEN 'custom' THEN 'custom' ELSE 'node' END`
  100. if err := tx.Exec(`UPDATE inbounds SET share_addr_strategy = ` + strategyExpr + ` WHERE share_addr_strategy IS NULL OR share_addr_strategy <> ` + strategyExpr).Error; err != nil {
  101. return err
  102. }
  103. hasShareAddr := tx.Migrator().HasColumn(&model.Inbound{}, "share_addr")
  104. if hasShareAddr {
  105. if err := tx.Exec(`UPDATE inbounds SET share_addr = TRIM(share_addr) WHERE share_addr IS NOT NULL AND share_addr <> TRIM(share_addr)`).Error; err != nil {
  106. return err
  107. }
  108. }
  109. if !hasShareAddr {
  110. return nil
  111. }
  112. var rows []struct {
  113. Id int
  114. ShareAddrStrategy string
  115. ShareAddr string
  116. }
  117. if err := tx.Model(&model.Inbound{}).Select("id", "share_addr_strategy", "share_addr").Find(&rows).Error; err != nil {
  118. return err
  119. }
  120. for _, row := range rows {
  121. strategy := normalizeInboundShareAddrStrategy(row.ShareAddrStrategy)
  122. addr, addrErr := normalizeInboundShareHost(row.ShareAddr)
  123. if addrErr != nil {
  124. strategy = "node"
  125. addr = ""
  126. }
  127. updates := map[string]any{}
  128. if strategy != row.ShareAddrStrategy {
  129. updates["share_addr_strategy"] = strategy
  130. }
  131. if addr != row.ShareAddr {
  132. updates["share_addr"] = addr
  133. }
  134. if len(updates) > 0 {
  135. if err := tx.Model(&model.Inbound{}).Where("id = ?", row.Id).Updates(updates).Error; err != nil {
  136. return err
  137. }
  138. }
  139. }
  140. return nil
  141. }
  142. // GetInbounds retrieves all inbounds for a specific user with client stats.
  143. func (s *InboundService) GetInbounds(userId int) ([]*model.Inbound, error) {
  144. db := database.GetDB()
  145. var inbounds []*model.Inbound
  146. err := db.Model(model.Inbound{}).Preload("ClientStats").Where("user_id = ?", userId).Order("id ASC").Find(&inbounds).Error
  147. if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
  148. return nil, err
  149. }
  150. s.enrichClientStats(db, inbounds)
  151. s.annotateFallbackParents(db, inbounds)
  152. s.annotateLocalOriginGuid(inbounds)
  153. return inbounds, nil
  154. }
  155. // annotateLocalOriginGuid fills OriginNodeGuid for this panel's OWN inbounds
  156. // (NodeID == nil) with the panel's stable GUID; inbounds synced from a node
  157. // already carry the originating node's GUID. Read-time only (not persisted) so
  158. // the per-inbound online view can scope by GUID uniformly across a chain of
  159. // nodes (#4983).
  160. func (s *InboundService) annotateLocalOriginGuid(inbounds []*model.Inbound) {
  161. if len(inbounds) == 0 {
  162. return
  163. }
  164. guid := s.panelGuid()
  165. if guid == "" {
  166. return
  167. }
  168. for _, ib := range inbounds {
  169. if ib.OriginNodeGuid == "" && ib.NodeID == nil {
  170. ib.OriginNodeGuid = guid
  171. }
  172. }
  173. }
  174. // GetInboundsSlim returns the same list of inbounds as GetInbounds but
  175. // strips every per-client field other than email / enable / comment from
  176. // settings.clients and skips UUID/SubId enrichment on ClientStats. The
  177. // inbounds page only needs those three to roll up client counts and
  178. // render badges, so this trims tens of bytes per client (UUID, password,
  179. // flow, security, totalGB, expiryTime, limitIp, tgId, ...) which adds
  180. // up fast on installs with thousands of clients.
  181. //
  182. // Full client data is still available through GET /panel/api/inbounds/get/:id
  183. // for the edit/info/qr/export/clone flows that need it.
  184. func (s *InboundService) GetInboundsSlim(userId int) ([]*model.Inbound, error) {
  185. db := database.GetDB()
  186. var inbounds []*model.Inbound
  187. err := db.Model(model.Inbound{}).Preload("ClientStats").Where("user_id = ?", userId).Order("id ASC").Find(&inbounds).Error
  188. if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
  189. return nil, err
  190. }
  191. s.annotateFallbackParents(db, inbounds)
  192. s.annotateLocalOriginGuid(inbounds)
  193. // Top up stats rows owned by sibling inbounds (multi-attached clients)
  194. // so the list's depleted/expiring badges see every client; the UUID/SubId
  195. // enrichment stays skipped. Must run before slimming strips the settings.
  196. s.backfillClientStats(db, inbounds)
  197. // Slim feeds the panel UI only (masters poll the full list), so the badge
  198. // math may see the cross-panel totals a master pushed.
  199. s.overlayInboundsClientStats(db, inbounds)
  200. for _, ib := range inbounds {
  201. ib.Settings = slimSettingsClients(ib.Settings)
  202. }
  203. return inbounds, nil
  204. }
  205. // slimSettingsClients rewrites the inbound settings JSON so settings.clients[]
  206. // keeps only the fields the list view actually reads. Returns the input
  207. // unchanged when the JSON can't be parsed or has no clients array.
  208. func slimSettingsClients(settings string) string {
  209. if settings == "" {
  210. return settings
  211. }
  212. var raw map[string]any
  213. if err := json.Unmarshal([]byte(settings), &raw); err != nil {
  214. return settings
  215. }
  216. clients, ok := raw["clients"].([]any)
  217. if !ok || len(clients) == 0 {
  218. return settings
  219. }
  220. slim := make([]any, 0, len(clients))
  221. for _, entry := range clients {
  222. c, ok := entry.(map[string]any)
  223. if !ok {
  224. continue
  225. }
  226. row := make(map[string]any, 3)
  227. if v, ok := c["email"]; ok {
  228. row["email"] = v
  229. }
  230. if v, ok := c["enable"]; ok {
  231. row["enable"] = v
  232. }
  233. if v, ok := c["comment"]; ok && v != "" {
  234. row["comment"] = v
  235. }
  236. slim = append(slim, row)
  237. }
  238. raw["clients"] = slim
  239. out, err := json.Marshal(raw)
  240. if err != nil {
  241. return settings
  242. }
  243. return string(out)
  244. }
  245. // annotateFallbackParents fills FallbackParent on each inbound that is
  246. // the child side of a fallback rule. One DB round-trip serves the full
  247. // list — the frontend needs this to rewrite the child's client-share
  248. // link so it points at the master's reachable endpoint.
  249. func (s *InboundService) annotateFallbackParents(db *gorm.DB, inbounds []*model.Inbound) {
  250. if len(inbounds) == 0 {
  251. return
  252. }
  253. childIds := make([]int, 0, len(inbounds))
  254. for _, ib := range inbounds {
  255. childIds = append(childIds, ib.Id)
  256. }
  257. var rows []model.InboundFallback
  258. if err := db.Where("child_id IN ?", childIds).
  259. Order("sort_order ASC, id ASC").
  260. Find(&rows).Error; err != nil {
  261. return
  262. }
  263. first := make(map[int]model.InboundFallback, len(rows))
  264. for _, r := range rows {
  265. if _, ok := first[r.ChildId]; !ok {
  266. first[r.ChildId] = r
  267. }
  268. }
  269. for _, ib := range inbounds {
  270. if r, ok := first[ib.Id]; ok {
  271. ib.FallbackParent = &model.FallbackParentInfo{
  272. MasterId: r.MasterId,
  273. Path: r.Path,
  274. }
  275. }
  276. }
  277. }
  278. type InboundOption struct {
  279. Id int `json:"id" example:"1"`
  280. Remark string `json:"remark" example:"VLESS-443"`
  281. Tag string `json:"tag" example:"in-443-tcp"`
  282. Protocol string `json:"protocol" example:"vless"`
  283. Port int `json:"port" example:"443"`
  284. Enable bool `json:"enable" example:"true"`
  285. TlsFlowCapable bool `json:"tlsFlowCapable" example:"true"`
  286. SsMethod string `json:"ssMethod"`
  287. WgPublicKey string `json:"wgPublicKey,omitempty"`
  288. WgMtu int `json:"wgMtu,omitempty"`
  289. WgDns string `json:"wgDns,omitempty"`
  290. MtprotoDomain string `json:"mtprotoDomain,omitempty"`
  291. // Hosting node; nil for this panel's own inbounds. Lets the clients
  292. // page map a node filter onto inbound IDs (#4997).
  293. NodeId *int `json:"nodeId,omitempty"`
  294. // Share-host resolution inputs, mirroring the subscription's
  295. // resolveInboundAddress so the clients page renders a node-managed WireGuard
  296. // Endpoint that points at the node, not the master panel. NodeAddress is the
  297. // hosting node's externally reachable address (empty for this panel's own
  298. // inbounds); Listen and ShareAddrStrategy/ShareAddr feed the same
  299. // node→listen→custom fallback the share/QR links already use.
  300. NodeAddress string `json:"nodeAddress,omitempty"`
  301. Listen string `json:"listen,omitempty"`
  302. ShareAddr string `json:"shareAddr,omitempty"`
  303. ShareAddrStrategy string `json:"shareAddrStrategy,omitempty"`
  304. }
  305. func (s *InboundService) GetInboundOptions(userId int) ([]InboundOption, error) {
  306. db := database.GetDB()
  307. var rows []struct {
  308. Id int `gorm:"column:id"`
  309. Remark string `gorm:"column:remark"`
  310. Tag string `gorm:"column:tag"`
  311. Protocol string `gorm:"column:protocol"`
  312. Port int `gorm:"column:port"`
  313. Enable bool `gorm:"column:enable"`
  314. StreamSettings string `gorm:"column:stream_settings"`
  315. Settings string `gorm:"column:settings"`
  316. Listen string `gorm:"column:listen"`
  317. ShareAddr string `gorm:"column:share_addr"`
  318. ShareAddrStrategy string `gorm:"column:share_addr_strategy"`
  319. NodeId *int `gorm:"column:node_id"`
  320. NodeAddress string `gorm:"column:node_address"`
  321. }
  322. err := db.Table("inbounds").
  323. Select("inbounds.id, inbounds.remark, inbounds.tag, inbounds.protocol, inbounds.port, inbounds.enable, inbounds.stream_settings, inbounds.settings, inbounds.listen, inbounds.share_addr, inbounds.share_addr_strategy, inbounds.node_id, COALESCE(nodes.address, '') AS node_address").
  324. Joins("LEFT JOIN nodes ON nodes.id = inbounds.node_id").
  325. Where("inbounds.user_id = ?", userId).
  326. Order("inbounds.id ASC").
  327. Scan(&rows).Error
  328. if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
  329. return nil, err
  330. }
  331. out := make([]InboundOption, 0, len(rows))
  332. for _, r := range rows {
  333. wgPublicKey, wgMtu, wgDns := inboundWireguardHints(r.Protocol, r.Settings)
  334. shareAddrStrategy := r.ShareAddrStrategy
  335. if shareAddrStrategy == "node" {
  336. shareAddrStrategy = ""
  337. }
  338. out = append(out, InboundOption{
  339. Id: r.Id,
  340. Remark: r.Remark,
  341. Tag: r.Tag,
  342. Protocol: r.Protocol,
  343. Port: r.Port,
  344. Enable: r.Enable,
  345. TlsFlowCapable: inboundCanEnableTlsFlow(r.Protocol, r.StreamSettings, r.Settings),
  346. SsMethod: inboundShadowsocksMethod(r.Protocol, r.Settings),
  347. WgPublicKey: wgPublicKey,
  348. WgMtu: wgMtu,
  349. WgDns: wgDns,
  350. MtprotoDomain: inboundMtprotoDomain(r.Protocol, r.Settings),
  351. NodeId: r.NodeId,
  352. NodeAddress: r.NodeAddress,
  353. Listen: r.Listen,
  354. ShareAddr: r.ShareAddr,
  355. ShareAddrStrategy: shareAddrStrategy,
  356. })
  357. }
  358. return out, nil
  359. }
  360. func inboundWireguardHints(protocol string, settings string) (string, int, string) {
  361. if protocol != string(model.WireGuard) || strings.TrimSpace(settings) == "" {
  362. return "", 0, ""
  363. }
  364. var parsed struct {
  365. PublicKey string `json:"publicKey"`
  366. PubKey string `json:"pubKey"`
  367. SecretKey string `json:"secretKey"`
  368. MTU int `json:"mtu"`
  369. DNS string `json:"dns"`
  370. }
  371. if err := json.Unmarshal([]byte(settings), &parsed); err != nil {
  372. return "", 0, ""
  373. }
  374. publicKey := parsed.PublicKey
  375. if publicKey == "" {
  376. publicKey = parsed.PubKey
  377. }
  378. if publicKey == "" && parsed.SecretKey != "" {
  379. if derived, err := wgutil.PublicKeyFromPrivate(parsed.SecretKey); err == nil {
  380. publicKey = derived
  381. }
  382. }
  383. return publicKey, parsed.MTU, parsed.DNS
  384. }
  385. // inboundMtprotoDomain returns the inbound-level FakeTLS default domain, used by
  386. // the clients UI to seed a new mtproto client's secret with the right fronting
  387. // hostname.
  388. func inboundMtprotoDomain(protocol string, settings string) string {
  389. if protocol != string(model.MTProto) || strings.TrimSpace(settings) == "" {
  390. return ""
  391. }
  392. var parsed struct {
  393. FakeTLSDomain string `json:"fakeTlsDomain"`
  394. }
  395. if err := json.Unmarshal([]byte(settings), &parsed); err != nil {
  396. return ""
  397. }
  398. return strings.TrimSpace(parsed.FakeTLSDomain)
  399. }
  400. // GetAllInbounds retrieves all inbounds with client stats.
  401. func (s *InboundService) GetAllInbounds() ([]*model.Inbound, error) {
  402. db := database.GetDB()
  403. var inbounds []*model.Inbound
  404. err := db.Model(model.Inbound{}).Preload("ClientStats").Find(&inbounds).Error
  405. if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
  406. return nil, err
  407. }
  408. s.enrichClientStats(db, inbounds)
  409. return inbounds, nil
  410. }
  411. func (s *InboundService) GetInboundsByTrafficReset(period string) ([]*model.Inbound, error) {
  412. db := database.GetDB()
  413. var inbounds []*model.Inbound
  414. err := db.Model(model.Inbound{}).Where("traffic_reset = ?", period).Find(&inbounds).Error
  415. if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
  416. return nil, err
  417. }
  418. return inbounds, nil
  419. }
  420. func (s *InboundService) GetClients(inbound *model.Inbound) ([]model.Client, error) {
  421. settings := map[string][]model.Client{}
  422. _ = json.Unmarshal([]byte(inbound.Settings), &settings)
  423. if settings == nil {
  424. return nil, fmt.Errorf("setting is null")
  425. }
  426. clients := settings["clients"]
  427. if clients == nil {
  428. return nil, nil
  429. }
  430. return clients, nil
  431. }
  432. // GetClientsBySubId returns the inbound's clients with the given subscription
  433. // id, resolved from the normalized clients tables (the same source the running
  434. // Xray users are built from) instead of parsing the settings JSON blob.
  435. func (s *InboundService) GetClientsBySubId(inboundId int, subId string) ([]model.Client, error) {
  436. return s.clientService.ListForInboundBySubId(nil, inboundId, subId)
  437. }
  438. func (s *InboundService) GetAllEmails() ([]string, error) {
  439. db := database.GetDB()
  440. var emails []string
  441. query := fmt.Sprintf(
  442. "SELECT DISTINCT %s %s",
  443. database.JSONFieldText("client.value", "email"),
  444. database.JSONClientsFromInbound(),
  445. )
  446. if err := db.Raw(query).Scan(&emails).Error; err != nil {
  447. return nil, err
  448. }
  449. return emails, nil
  450. }
  451. // getAllEmailSubIDs returns email→subId. An email seen with two different
  452. // non-empty subIds is locked (mapped to "") so neither identity can claim it.
  453. func (s *InboundService) getAllEmailSubIDs() (map[string]string, error) {
  454. db := database.GetDB()
  455. var rows []struct {
  456. Email string
  457. SubID string
  458. }
  459. query := fmt.Sprintf(
  460. "SELECT %s AS email, %s AS sub_id %s",
  461. database.JSONFieldText("client.value", "email"),
  462. database.JSONFieldText("client.value", "subId"),
  463. database.JSONClientsFromInbound(),
  464. )
  465. if err := db.Raw(query).Scan(&rows).Error; err != nil {
  466. return nil, err
  467. }
  468. result := make(map[string]string, len(rows))
  469. for _, r := range rows {
  470. email := strings.ToLower(r.Email)
  471. if email == "" {
  472. continue
  473. }
  474. subID := r.SubID
  475. if existing, ok := result[email]; ok {
  476. if existing != subID {
  477. result[email] = ""
  478. }
  479. continue
  480. }
  481. result[email] = subID
  482. }
  483. return result, nil
  484. }
  485. // normalizeStreamSettings clears StreamSettings for protocols that don't use it.
  486. // Only vmess, vless, trojan, shadowsocks, hysteria, wireguard, and tunnel
  487. // protocols use streamSettings (wireguard for finalmask UDP masks and sockopt on
  488. // its listener; tunnel for sockopt, notably sockopt.tproxy for its TProxy/redirect
  489. // mode).
  490. func (s *InboundService) normalizeStreamSettings(inbound *model.Inbound) {
  491. protocolsWithStream := map[model.Protocol]bool{
  492. model.VMESS: true,
  493. model.VLESS: true,
  494. model.Trojan: true,
  495. model.Shadowsocks: true,
  496. model.Hysteria: true,
  497. model.WireGuard: true,
  498. model.Tunnel: true,
  499. }
  500. if !protocolsWithStream[inbound.Protocol] {
  501. inbound.StreamSettings = ""
  502. }
  503. }
  504. // normalizeMtprotoSecret rebuilds every mtproto client's FakeTLS secret so it is
  505. // always valid before the row is persisted, and drops the vestigial inbound-level
  506. // secret and adTag: MTProto is multi-client, so mtg and every share link read
  507. // only the per-client values. Leaving an inbound-level secret behind is what
  508. // produced stale links that failed with "incorrect client random".
  509. func (s *InboundService) normalizeMtprotoSecret(inbound *model.Inbound) {
  510. if inbound.Protocol != model.MTProto {
  511. return
  512. }
  513. if stripped, ok := model.StripMtprotoInboundSecret(inbound.Settings); ok {
  514. inbound.Settings = stripped
  515. }
  516. if stripped, ok := model.StripMtprotoInboundAdTag(inbound.Settings); ok {
  517. inbound.Settings = stripped
  518. }
  519. if healed, ok := model.HealMtprotoClientSecrets(inbound.Settings); ok {
  520. inbound.Settings = healed
  521. }
  522. }
  523. // mtprotoRoutesThroughXray reports whether an mtproto inbound is configured to
  524. // egress through the core's router (the loopback SOCKS bridge in §xray.go).
  525. func mtprotoRoutesThroughXray(inbound *model.Inbound) bool {
  526. if inbound == nil || inbound.Protocol != model.MTProto {
  527. return false
  528. }
  529. var parsed struct {
  530. RouteThroughXray bool `json:"routeThroughXray"`
  531. }
  532. if err := json.Unmarshal([]byte(inbound.Settings), &parsed); err != nil {
  533. return false
  534. }
  535. return parsed.RouteThroughXray
  536. }
  537. func settingsRouteXrayPort(parsed map[string]any) int {
  538. switch v := parsed["routeXrayPort"].(type) {
  539. case float64:
  540. return int(v)
  541. case int:
  542. return v
  543. case json.Number:
  544. if n, err := v.Int64(); err == nil {
  545. return int(n)
  546. }
  547. }
  548. return 0
  549. }
  550. func parseRouteXrayPort(settings string) int {
  551. if settings == "" {
  552. return 0
  553. }
  554. var parsed map[string]any
  555. if err := json.Unmarshal([]byte(settings), &parsed); err != nil {
  556. return 0
  557. }
  558. return settingsRouteXrayPort(parsed)
  559. }
  560. // normalizeMtprotoXrayPort guarantees a routed mtproto inbound carries a stable
  561. // loopback egress port in its settings, so the generated Xray SOCKS bridge and
  562. // the mtg sidecar agree on where mtg dials out. The port is backend-owned: it is
  563. // allocated once when routing is first enabled and preserved across edits
  564. // (carried over from oldSettings, which wins over any value the client echoed
  565. // back). When routing is off it — together with the now-inert outbound
  566. // selection — is stripped so a disabled bridge leaves nothing stale behind.
  567. //
  568. // It returns an error when an egress port cannot be allocated or persisted, so
  569. // the caller refuses the save rather than storing a routed-but-portless inbound,
  570. // which would otherwise route no traffic and have its mtg metrics skipped (see
  571. // mtproto_job) — silently losing its accounting.
  572. func (s *InboundService) normalizeMtprotoXrayPort(inbound *model.Inbound, oldSettings string) error {
  573. if inbound.Protocol != model.MTProto {
  574. return nil
  575. }
  576. var parsed map[string]any
  577. if err := json.Unmarshal([]byte(inbound.Settings), &parsed); err != nil || parsed == nil {
  578. return nil
  579. }
  580. routed, _ := parsed["routeThroughXray"].(bool)
  581. if !routed {
  582. _, hadPort := parsed["routeXrayPort"]
  583. _, hadTag := parsed["outboundTag"]
  584. if !hadPort && !hadTag {
  585. return nil
  586. }
  587. delete(parsed, "routeXrayPort")
  588. delete(parsed, "outboundTag")
  589. if bs, err := json.MarshalIndent(parsed, "", " "); err == nil {
  590. inbound.Settings = string(bs)
  591. } else {
  592. logger.Warning("mtproto: failed to marshal settings after disabling routing:", err)
  593. }
  594. return nil
  595. }
  596. // Prefer the already-stored port (carried across edits), then any value the
  597. // client sent, then allocate a fresh one.
  598. port := parseRouteXrayPort(oldSettings)
  599. if port <= 0 {
  600. port = settingsRouteXrayPort(parsed)
  601. }
  602. if port <= 0 {
  603. allocated, err := mtproto.FreeLocalPort()
  604. if err != nil {
  605. return common.NewError("mtproto: could not allocate an Xray egress port:", err)
  606. }
  607. port = allocated
  608. }
  609. if settingsRouteXrayPort(parsed) == port {
  610. return nil
  611. }
  612. parsed["routeXrayPort"] = port
  613. bs, err := json.MarshalIndent(parsed, "", " ")
  614. if err != nil {
  615. return common.NewError("mtproto: could not persist the Xray egress port:", err)
  616. }
  617. inbound.Settings = string(bs)
  618. return nil
  619. }
  620. // AddInbound creates a new inbound configuration.
  621. // It validates port uniqueness, client email uniqueness, and required fields,
  622. // then saves the inbound to the database and optionally adds it to the running Xray instance.
  623. // Returns the created inbound, whether Xray needs restart, and any error.
  624. func (s *InboundService) AddInbound(inbound *model.Inbound) (*model.Inbound, bool, error) {
  625. // Normalize streamSettings based on protocol
  626. s.normalizeStreamSettings(inbound)
  627. s.normalizeMtprotoSecret(inbound)
  628. if err := s.normalizeMtprotoXrayPort(inbound, ""); err != nil {
  629. return inbound, false, err
  630. }
  631. inbound.SubSortIndex = normalizeSubSortIndex(inbound.SubSortIndex)
  632. if err := normalizeInboundShareAddressStrict(inbound); err != nil {
  633. return inbound, false, err
  634. }
  635. conflict, err := s.checkPortConflict(inbound, 0)
  636. if err != nil {
  637. return inbound, false, err
  638. }
  639. if conflict != nil {
  640. return inbound, false, common.NewError(conflict.String())
  641. }
  642. inbound.Tag, err = s.resolveInboundTag(inbound, 0)
  643. if err != nil {
  644. return inbound, false, err
  645. }
  646. clients, err := s.GetClients(inbound)
  647. if err != nil {
  648. return inbound, false, err
  649. }
  650. existEmail, err := s.clientService.checkEmailsExistForClients(s, clients, nil)
  651. if err != nil {
  652. return inbound, false, err
  653. }
  654. if existEmail != "" {
  655. return inbound, false, common.NewError("Duplicate email:", existEmail)
  656. }
  657. // Ensure created_at and updated_at on clients in settings
  658. if len(clients) > 0 {
  659. var settings map[string]any
  660. if err2 := json.Unmarshal([]byte(inbound.Settings), &settings); err2 == nil && settings != nil {
  661. now := time.Now().Unix() * 1000
  662. updatedClients := make([]model.Client, 0, len(clients))
  663. for _, c := range clients {
  664. if c.CreatedAt == 0 {
  665. c.CreatedAt = now
  666. }
  667. c.UpdatedAt = now
  668. updatedClients = append(updatedClients, c)
  669. }
  670. settings["clients"] = updatedClients
  671. if bs, err3 := json.MarshalIndent(settings, "", " "); err3 == nil {
  672. inbound.Settings = string(bs)
  673. } else {
  674. logger.Debug("Unable to marshal inbound settings with timestamps:", err3)
  675. }
  676. } else if err2 != nil {
  677. logger.Debug("Unable to parse inbound settings for timestamps:", err2)
  678. }
  679. }
  680. // Defensively fix any Shadowsocks-2022 client PSK whose length doesn't match
  681. // the inbound method (e.g. an API caller supplied a wrong-size key).
  682. if normalized, changed := normalizeShadowsocksClientKeys(inbound.Settings); changed {
  683. inbound.Settings = normalized
  684. }
  685. // Secure client ID
  686. for _, client := range clients {
  687. switch inbound.Protocol {
  688. case "trojan":
  689. if client.Password == "" {
  690. return inbound, false, common.NewError("empty client ID")
  691. }
  692. case "shadowsocks":
  693. if client.Email == "" {
  694. return inbound, false, common.NewError("empty client ID")
  695. }
  696. case "hysteria":
  697. if client.Auth == "" {
  698. return inbound, false, common.NewError("empty client ID")
  699. }
  700. case "mtproto":
  701. if client.Secret == "" {
  702. return inbound, false, common.NewError("mtproto client requires a secret")
  703. }
  704. if client.AdTag != "" && !model.ValidMtprotoAdTag(client.AdTag) {
  705. return inbound, false, common.NewError("mtproto client ad tag must be 32 hex characters")
  706. }
  707. default:
  708. if client.ID == "" {
  709. return inbound, false, common.NewError("empty client ID")
  710. }
  711. }
  712. }
  713. db := database.GetDB()
  714. tx := db.Begin()
  715. markDirty := false
  716. defer func() {
  717. if err != nil {
  718. tx.Rollback()
  719. return
  720. }
  721. if markDirty && inbound.NodeID != nil {
  722. if dErr := (&NodeService{}).MarkNodeDirtyTx(tx, *inbound.NodeID); dErr != nil {
  723. err = dErr
  724. tx.Rollback()
  725. return
  726. }
  727. }
  728. tx.Commit()
  729. }()
  730. // Omit the ClientStats has-many association: GORM's cascade would INSERT
  731. // those rows with an ON CONFLICT target on the primary key only, which
  732. // collides with the globally-unique client_traffics.email when an imported
  733. // inbound carries clients that another inbound already created (e.g.
  734. // importing two inbounds that share the same clients). We insert the stats
  735. // ourselves below with the same email-conflict guard AddClientStat uses.
  736. err = tx.Omit("ClientStats").Save(inbound).Error
  737. if err != nil {
  738. return inbound, false, err
  739. }
  740. // Imported stats first, so their traffic counters survive; emails that
  741. // already own a (shared) row are skipped instead of tripping the unique
  742. // constraint.
  743. for i := range inbound.ClientStats {
  744. if inbound.ClientStats[i].Email == "" {
  745. continue
  746. }
  747. inbound.ClientStats[i].Id = 0
  748. inbound.ClientStats[i].InboundId = inbound.Id
  749. if err = tx.Clauses(clause.OnConflict{
  750. Columns: []clause.Column{{Name: "email"}},
  751. DoNothing: true,
  752. }).Create(&inbound.ClientStats[i]).Error; err != nil {
  753. return inbound, false, err
  754. }
  755. }
  756. // Then make sure every client has a stats row. AddClientStat is a no-op
  757. // where one exists (including the rows just inserted), and fills the gap
  758. // for clients an import payload didn't carry stats for.
  759. for _, client := range clients {
  760. if err = s.AddClientStat(tx, inbound.Id, &client); err != nil {
  761. return inbound, false, err
  762. }
  763. }
  764. if err = s.clientService.SyncInbound(tx, inbound.Id, clients); err != nil {
  765. return inbound, false, err
  766. }
  767. // Legacy import: an inbound exported from a build that predated the hosts
  768. // table carries its external proxies inline in streamSettings.externalProxy.
  769. // The startup migration that converts those to host rows runs once and is
  770. // gated off afterwards, so it never sees a freshly imported inbound —
  771. // reproduce it here. No-op for inbounds without externalProxy (everything the
  772. // current UI builds), so this only fires on such imports.
  773. if _, err = database.CreateHostsFromExternalProxy(tx, inbound.Id, inbound.StreamSettings); err != nil {
  774. return inbound, false, err
  775. }
  776. // Before the deferred commit, so a node in "selected" sync mode cannot
  777. // sweep the new central row in the gap before its tag is allowed.
  778. if inbound.NodeID != nil {
  779. if aErr := (&NodeService{}).EnsureInboundTagAllowed(*inbound.NodeID, inbound.Tag); aErr != nil {
  780. logger.Warning("allow inbound tag on node failed:", aErr)
  781. }
  782. }
  783. needRestart := false
  784. if inbound.Enable {
  785. rt, push, dirty, perr := s.nodePushPlan(inbound)
  786. if perr != nil {
  787. err = perr
  788. return inbound, false, err
  789. }
  790. if dirty {
  791. markDirty = true
  792. }
  793. if push {
  794. payload := inbound
  795. pushable := true
  796. if inbound.NodeID == nil && inbound.Protocol == model.MTProto {
  797. if built, bErr := s.buildRuntimeInboundForAPI(tx, inbound); bErr == nil {
  798. payload = built
  799. } else {
  800. logger.Debug("Unable to prepare runtime inbound config:", bErr)
  801. pushable = false
  802. }
  803. }
  804. if pushable {
  805. if err1 := rt.AddInbound(context.Background(), payload); err1 == nil {
  806. logger.Debug("New inbound added on", rt.Name(), ":", inbound.Tag)
  807. } else {
  808. logger.Debug("Unable to add inbound on", rt.Name(), ":", err1)
  809. if inbound.NodeID != nil {
  810. markDirty = true
  811. } else if inbound.Protocol != model.MTProto {
  812. needRestart = true
  813. }
  814. }
  815. }
  816. }
  817. }
  818. // A routed mtproto inbound is not an Xray inbound itself, so the runtime
  819. // push above only (re)starts the mtg sidecar. The egress SOCKS bridge lives
  820. // in the generated config, so force a regen to wire it in.
  821. if mtprotoRoutesThroughXray(inbound) {
  822. needRestart = true
  823. }
  824. return inbound, needRestart, err
  825. }
  826. func (s *InboundService) DelInbound(id int) (bool, error) {
  827. db := database.GetDB()
  828. needRestart := false
  829. markDirty := false
  830. var ib model.Inbound
  831. loadErr := db.Model(model.Inbound{}).Where("id = ?", id).First(&ib).Error
  832. if loadErr == nil {
  833. shouldPushToRuntime := ib.NodeID != nil || ib.Enable
  834. if shouldPushToRuntime {
  835. rt, push, dirty, perr := s.nodePushPlan(&ib)
  836. if perr != nil {
  837. logger.Warning("DelInbound: node lookup failed, deleting central row anyway:", perr)
  838. markDirty = true
  839. } else if push {
  840. if err1 := rt.DelInbound(context.Background(), &ib); err1 == nil {
  841. logger.Debug("Inbound deleted on", rt.Name(), ":", ib.Tag)
  842. } else {
  843. logger.Warning("DelInbound on", rt.Name(), "failed, deleting central row anyway:", err1)
  844. if ib.NodeID == nil {
  845. needRestart = true
  846. } else {
  847. markDirty = true
  848. }
  849. }
  850. } else if ib.NodeID == nil {
  851. needRestart = true
  852. } else if dirty {
  853. markDirty = true
  854. }
  855. } else {
  856. logger.Debug("DelInbound: skipping runtime push for disabled local inbound id:", id)
  857. }
  858. } else {
  859. logger.Debug("DelInbound: inbound not found, id:", id)
  860. }
  861. if err := s.clientService.DetachInbound(db, id); err != nil {
  862. return false, err
  863. }
  864. // Drop the deleted inbound's tag from any routing rules / loopback outbounds
  865. // in xrayTemplateConfig so they don't point at a tag that no longer exists.
  866. if loadErr == nil && ib.Tag != "" {
  867. if routingChanged, syncErr := (&XraySettingService{}).RemoveInboundTagReferences(ib.Tag); syncErr != nil {
  868. logger.Warning("DelInbound: sync routing on inbound delete failed:", syncErr)
  869. } else if routingChanged {
  870. needRestart = true
  871. }
  872. }
  873. if err := db.Transaction(func(tx *gorm.DB) error {
  874. if err := tx.Delete(model.Inbound{}, id).Error; err != nil {
  875. return err
  876. }
  877. // Hosts have no hard FK; drop the inbound's hosts alongside it.
  878. if err := tx.Where("inbound_id = ?", id).Delete(&model.Host{}).Error; err != nil {
  879. return err
  880. }
  881. if markDirty && ib.NodeID != nil {
  882. return (&NodeService{}).MarkNodeDirtyTx(tx, *ib.NodeID)
  883. }
  884. return nil
  885. }); err != nil {
  886. return needRestart, err
  887. }
  888. if !database.IsPostgres() {
  889. var count int64
  890. if err := db.Model(&model.Inbound{}).Count(&count).Error; err != nil {
  891. return needRestart, err
  892. }
  893. if count == 0 {
  894. if err := db.Exec("DELETE FROM sqlite_sequence WHERE name = ?", "inbounds").Error; err != nil {
  895. return needRestart, err
  896. }
  897. }
  898. }
  899. // Drop the egress SOCKS bridge a routed mtproto inbound left in the config.
  900. if mtprotoRoutesThroughXray(&ib) {
  901. needRestart = true
  902. }
  903. return needRestart, nil
  904. }
  905. type BulkDelInboundResult struct {
  906. Deleted int `json:"deleted"`
  907. Skipped []BulkDelInboundReport `json:"skipped,omitempty"`
  908. }
  909. type BulkDelInboundReport struct {
  910. Id int `json:"id"`
  911. Reason string `json:"reason"`
  912. }
  913. // DelInbounds removes every inbound in the list, reusing the single-delete
  914. // path per id. Failures are recorded in Skipped and processing continues for
  915. // the rest; the aggregated needRestart is returned so the caller restarts
  916. // xray at most once.
  917. func (s *InboundService) DelInbounds(ids []int) (BulkDelInboundResult, bool, error) {
  918. result := BulkDelInboundResult{}
  919. needRestart := false
  920. for _, id := range ids {
  921. r, err := s.DelInbound(id)
  922. if err != nil {
  923. result.Skipped = append(result.Skipped, BulkDelInboundReport{Id: id, Reason: err.Error()})
  924. continue
  925. }
  926. result.Deleted++
  927. if r {
  928. needRestart = true
  929. }
  930. }
  931. return result, needRestart, nil
  932. }
  933. func (s *InboundService) GetInbound(id int) (*model.Inbound, error) {
  934. db := database.GetDB()
  935. inbound := &model.Inbound{}
  936. err := db.Model(model.Inbound{}).First(inbound, id).Error
  937. if err != nil {
  938. return nil, err
  939. }
  940. return inbound, nil
  941. }
  942. func (s *InboundService) GetInboundDetail(id int) (*model.Inbound, error) {
  943. db := database.GetDB()
  944. inbound := &model.Inbound{}
  945. err := db.Model(model.Inbound{}).Preload("ClientStats").First(inbound, id).Error
  946. if err != nil {
  947. return nil, err
  948. }
  949. s.enrichClientStats(db, []*model.Inbound{inbound})
  950. s.overlayInboundsClientStats(db, []*model.Inbound{inbound})
  951. return inbound, nil
  952. }
  953. func (s *InboundService) SetInboundEnable(id int, enable bool) (bool, error) {
  954. inbound, err := s.GetInbound(id)
  955. if err != nil {
  956. return false, err
  957. }
  958. if inbound.Enable == enable {
  959. return false, nil
  960. }
  961. db := database.GetDB()
  962. if err := db.Transaction(func(tx *gorm.DB) error {
  963. if err := tx.Model(model.Inbound{}).Where("id = ?", id).
  964. Update("enable", enable).Error; err != nil {
  965. return err
  966. }
  967. if inbound.NodeID != nil {
  968. return (&NodeService{}).MarkNodeDirtyTx(tx, *inbound.NodeID)
  969. }
  970. return nil
  971. }); err != nil {
  972. return false, err
  973. }
  974. inbound.Enable = enable
  975. needRestart := false
  976. rt, push, _, perr := s.nodePushPlan(inbound)
  977. if perr != nil {
  978. return false, perr
  979. }
  980. // Remote nodes interpret DelInbound as a real row delete (it hits
  981. // panel/api/inbounds/del/:id on the remote), so toggling the enable
  982. // switch on a remote inbound used to wipe the row entirely (#4402).
  983. // PATCH the remote row via UpdateInbound instead — preserves the
  984. // settings/client history and just flips the enable flag.
  985. if inbound.NodeID != nil {
  986. if push {
  987. if err := rt.UpdateInbound(context.Background(), inbound, inbound); err != nil {
  988. logger.Warning("SetInboundEnable: remote UpdateInbound on", rt.Name(), "failed:", err)
  989. }
  990. }
  991. return false, nil
  992. }
  993. if !push {
  994. return true, nil
  995. }
  996. if err := rt.DelInbound(context.Background(), inbound); err != nil &&
  997. !strings.Contains(err.Error(), "not found") {
  998. logger.Debug("SetInboundEnable: DelInbound on", rt.Name(), "failed:", err)
  999. needRestart = true
  1000. }
  1001. if !enable {
  1002. return needRestart, nil
  1003. }
  1004. runtimeInbound, err := s.buildRuntimeInboundForAPI(db, inbound)
  1005. if err != nil {
  1006. logger.Debug("SetInboundEnable: build runtime config failed:", err)
  1007. return true, nil
  1008. }
  1009. if err := rt.AddInbound(context.Background(), runtimeInbound); err != nil {
  1010. logger.Debug("SetInboundEnable: AddInbound on", rt.Name(), "failed:", err)
  1011. needRestart = true
  1012. }
  1013. return needRestart, nil
  1014. }
  1015. func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound, bool, error) {
  1016. // Normalize streamSettings based on protocol
  1017. s.normalizeStreamSettings(inbound)
  1018. s.normalizeMtprotoSecret(inbound)
  1019. inbound.SubSortIndex = normalizeSubSortIndex(inbound.SubSortIndex)
  1020. conflict, err := s.checkPortConflict(inbound, inbound.Id)
  1021. if err != nil {
  1022. return inbound, false, err
  1023. }
  1024. if conflict != nil {
  1025. return inbound, false, common.NewError(conflict.String())
  1026. }
  1027. oldInbound, err := s.GetInbound(inbound.Id)
  1028. if err != nil {
  1029. return inbound, false, err
  1030. }
  1031. inbound.NodeID = oldInbound.NodeID
  1032. // Capture the pre-edit protocol and routing state before oldInbound is
  1033. // overwritten with the new values further down, then ensure a routed
  1034. // inbound keeps a stable egress port (reusing the one already stored).
  1035. oldProtocol := oldInbound.Protocol
  1036. oldRoutedMtproto := mtprotoRoutesThroughXray(oldInbound)
  1037. if err := s.normalizeMtprotoXrayPort(inbound, oldInbound.Settings); err != nil {
  1038. return inbound, false, err
  1039. }
  1040. tag := oldInbound.Tag
  1041. oldBits := inboundTransports(oldInbound.Protocol, oldInbound.StreamSettings, oldInbound.Settings)
  1042. oldTagWasAuto := isAutoGeneratedTag(tag, oldInbound.Port, oldInbound.NodeID, oldBits)
  1043. needRestart := false
  1044. // Persist the client-stat sync, settings munging, runtime push and inbound
  1045. // save as one transaction routed through the serial traffic writer, so it
  1046. // never runs concurrently with the @every 5s traffic poll. Both touch
  1047. // client_traffics and inbounds in opposite order, which Postgres aborts as a
  1048. // deadlock (40P01); serializing removes the contention (runSerializedTx).
  1049. //
  1050. // The runtime push stays inside the transaction here (unlike the client-edit
  1051. // paths that apply it after commit): EnsureInboundTagAllowed must reach the
  1052. // node before the central row is committed, or a "selected"-mode node would
  1053. // sweep the renamed inbound on its next pull. Inbound edits are rare, so
  1054. // holding the writer across the node call is an acceptable trade.
  1055. txErr := runSerializedTx(func(tx *gorm.DB) error {
  1056. if err := s.updateClientTraffics(tx, oldInbound, inbound); err != nil {
  1057. return err
  1058. }
  1059. // Ensure created_at and updated_at exist in inbound.Settings clients
  1060. {
  1061. var oldSettings map[string]any
  1062. _ = json.Unmarshal([]byte(oldInbound.Settings), &oldSettings)
  1063. emailToCreated := map[string]int64{}
  1064. emailToUpdated := map[string]int64{}
  1065. if oldSettings != nil {
  1066. if oc, ok := oldSettings["clients"].([]any); ok {
  1067. for _, it := range oc {
  1068. if m, ok2 := it.(map[string]any); ok2 {
  1069. if email, ok3 := m["email"].(string); ok3 {
  1070. switch v := m["created_at"].(type) {
  1071. case float64:
  1072. emailToCreated[email] = int64(v)
  1073. case int64:
  1074. emailToCreated[email] = v
  1075. }
  1076. switch v := m["updated_at"].(type) {
  1077. case float64:
  1078. emailToUpdated[email] = int64(v)
  1079. case int64:
  1080. emailToUpdated[email] = v
  1081. }
  1082. }
  1083. }
  1084. }
  1085. }
  1086. }
  1087. var newSettings map[string]any
  1088. if err2 := json.Unmarshal([]byte(inbound.Settings), &newSettings); err2 == nil && newSettings != nil {
  1089. now := time.Now().Unix() * 1000
  1090. if nSlice, ok := newSettings["clients"].([]any); ok {
  1091. for i := range nSlice {
  1092. if m, ok2 := nSlice[i].(map[string]any); ok2 {
  1093. email, _ := m["email"].(string)
  1094. if _, ok3 := m["created_at"]; !ok3 {
  1095. if v, ok4 := emailToCreated[email]; ok4 && v > 0 {
  1096. m["created_at"] = v
  1097. } else {
  1098. m["created_at"] = now
  1099. }
  1100. }
  1101. // Preserve client's updated_at if present; do not bump on parent inbound update
  1102. if _, hasUpdated := m["updated_at"]; !hasUpdated {
  1103. if v, ok4 := emailToUpdated[email]; ok4 && v > 0 {
  1104. m["updated_at"] = v
  1105. }
  1106. }
  1107. nSlice[i] = m
  1108. }
  1109. }
  1110. newSettings["clients"] = nSlice
  1111. if bs, err3 := json.MarshalIndent(newSettings, "", " "); err3 == nil {
  1112. inbound.Settings = string(bs)
  1113. }
  1114. }
  1115. }
  1116. }
  1117. // A Shadowsocks-2022 method change resizes the key, but existing client PSKs
  1118. // keep their old length and would be rejected by xray. Regenerate mismatched
  1119. // client keys so the inbound stays connectable.
  1120. if normalized, changed := normalizeShadowsocksClientKeys(inbound.Settings); changed {
  1121. inbound.Settings = normalized
  1122. logger.Warning("Shadowsocks inbound", inbound.Id, "method change resized keys; regenerated mismatched client PSK(s)")
  1123. }
  1124. // Re-gate Vision flow now that the new stream/encryption is known: if this
  1125. // VLESS inbound just became flow-eligible (e.g. vlessenc was enabled on an
  1126. // XHTTP inbound), restore Vision for clients whose intended flow is Vision
  1127. // but was stripped while the inbound was ineligible.
  1128. if restored, changed := s.restoreVisionFlowForEligibleInbound(tx, inbound.Settings, inbound.StreamSettings, inbound.Protocol); changed {
  1129. inbound.Settings = restored
  1130. }
  1131. oldInbound.Total = inbound.Total
  1132. oldInbound.Remark = inbound.Remark
  1133. oldInbound.SubSortIndex = inbound.SubSortIndex
  1134. oldInbound.Enable = inbound.Enable
  1135. oldInbound.ExpiryTime = inbound.ExpiryTime
  1136. oldInbound.TrafficReset = inbound.TrafficReset
  1137. oldInbound.Listen = inbound.Listen
  1138. oldInbound.Port = inbound.Port
  1139. oldInbound.Protocol = inbound.Protocol
  1140. oldInbound.Settings = inbound.Settings
  1141. oldInbound.StreamSettings = inbound.StreamSettings
  1142. oldInbound.Sniffing = inbound.Sniffing
  1143. if strings.TrimSpace(inbound.ShareAddrStrategy) == "" {
  1144. normalizeInboundShareAddress(oldInbound)
  1145. inbound.ShareAddrStrategy = oldInbound.ShareAddrStrategy
  1146. inbound.ShareAddr = oldInbound.ShareAddr
  1147. } else {
  1148. if err := normalizeInboundShareAddressStrict(inbound); err != nil {
  1149. return err
  1150. }
  1151. oldInbound.ShareAddrStrategy = inbound.ShareAddrStrategy
  1152. oldInbound.ShareAddr = inbound.ShareAddr
  1153. }
  1154. if oldTagWasAuto && inbound.Tag == tag {
  1155. inbound.Tag = ""
  1156. }
  1157. resolvedTag, err := s.resolveInboundTag(inbound, inbound.Id)
  1158. if err != nil {
  1159. return err
  1160. }
  1161. oldInbound.Tag = resolvedTag
  1162. inbound.Tag = oldInbound.Tag
  1163. rt, push, _, perr := s.nodePushPlan(oldInbound)
  1164. if perr != nil {
  1165. return perr
  1166. }
  1167. if oldInbound.NodeID == nil {
  1168. if !push {
  1169. needRestart = true
  1170. } else if oldProtocol == model.MTProto || oldInbound.Protocol == model.MTProto {
  1171. oldSnapshot := *oldInbound
  1172. oldSnapshot.Tag = tag
  1173. oldSnapshot.Protocol = oldProtocol
  1174. payload := oldInbound
  1175. pushable := true
  1176. if inbound.Enable {
  1177. if built, err2 := s.buildRuntimeInboundForAPI(tx, oldInbound); err2 == nil {
  1178. payload = built
  1179. } else {
  1180. logger.Debug("Unable to prepare runtime inbound config:", err2)
  1181. pushable = false
  1182. }
  1183. }
  1184. if pushable {
  1185. if err2 := rt.UpdateInbound(context.Background(), &oldSnapshot, payload); err2 == nil {
  1186. logger.Debug("Updated inbound applied on", rt.Name(), ":", oldInbound.Tag)
  1187. } else {
  1188. logger.Debug("Unable to update inbound on", rt.Name(), ":", err2)
  1189. if oldInbound.Protocol != model.MTProto {
  1190. needRestart = true
  1191. }
  1192. }
  1193. }
  1194. } else {
  1195. oldSnapshot := *oldInbound
  1196. oldSnapshot.Tag = tag
  1197. if err2 := rt.DelInbound(context.Background(), &oldSnapshot); err2 == nil {
  1198. logger.Debug("Old inbound deleted on", rt.Name(), ":", tag)
  1199. }
  1200. if inbound.Enable {
  1201. runtimeInbound, err2 := s.buildRuntimeInboundForAPI(tx, oldInbound)
  1202. if err2 != nil {
  1203. logger.Debug("Unable to prepare runtime inbound config:", err2)
  1204. needRestart = true
  1205. } else if err2 := rt.AddInbound(context.Background(), runtimeInbound); err2 == nil {
  1206. logger.Debug("Updated inbound added on", rt.Name(), ":", oldInbound.Tag)
  1207. } else {
  1208. logger.Debug("Unable to update inbound on", rt.Name(), ":", err2)
  1209. needRestart = true
  1210. }
  1211. }
  1212. }
  1213. } else if push {
  1214. oldSnapshot := *oldInbound
  1215. oldSnapshot.Tag = tag
  1216. if !inbound.Enable {
  1217. if err2 := rt.DelInbound(context.Background(), &oldSnapshot); err2 != nil {
  1218. logger.Warning("Unable to disable inbound on", rt.Name(), ":", err2)
  1219. }
  1220. } else if err2 := rt.UpdateInbound(context.Background(), &oldSnapshot, oldInbound); err2 != nil {
  1221. logger.Warning("Unable to update inbound on", rt.Name(), ":", err2)
  1222. }
  1223. }
  1224. // A rename must allow the new tag before the inbound row is committed, or a
  1225. // node in "selected" sync mode would sweep the renamed central row on the
  1226. // next pull.
  1227. if oldInbound.NodeID != nil {
  1228. if aErr := (&NodeService{}).EnsureInboundTagAllowed(*oldInbound.NodeID, oldInbound.Tag); aErr != nil {
  1229. logger.Warning("allow inbound tag on node failed:", aErr)
  1230. }
  1231. }
  1232. if err := tx.Save(oldInbound).Error; err != nil {
  1233. return err
  1234. }
  1235. newClients, gcErr := s.GetClients(oldInbound)
  1236. if gcErr != nil {
  1237. return gcErr
  1238. }
  1239. if err := s.clientService.SyncInbound(tx, oldInbound.Id, newClients); err != nil {
  1240. return err
  1241. }
  1242. if oldInbound.NodeID != nil {
  1243. if err := (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID); err != nil {
  1244. return err
  1245. }
  1246. }
  1247. // (Re)generate the Xray config whenever routing was or is now enabled, so
  1248. // the egress SOCKS bridge is added, moved, or dropped to match the new
  1249. // settings.
  1250. if mtprotoRoutesThroughXray(inbound) || oldRoutedMtproto {
  1251. needRestart = true
  1252. }
  1253. return nil
  1254. })
  1255. if txErr != nil {
  1256. return inbound, false, txErr
  1257. }
  1258. // After the rename is committed, point any routing rules / loopback outbounds
  1259. // in xrayTemplateConfig at the new tag (oldInbound.Tag now holds the resolved
  1260. // new tag; tag holds the pre-edit one). Done post-commit so a sync failure
  1261. // can't roll back the inbound edit.
  1262. if tag != oldInbound.Tag {
  1263. if routingChanged, syncErr := (&XraySettingService{}).PropagateInboundTagRename(tag, oldInbound.Tag); syncErr != nil {
  1264. logger.Warning("UpdateInbound: sync routing on tag rename failed:", syncErr)
  1265. } else if routingChanged {
  1266. needRestart = true
  1267. }
  1268. }
  1269. return inbound, needRestart, nil
  1270. }
  1271. func (s *InboundService) buildRuntimeInboundForAPI(tx *gorm.DB, inbound *model.Inbound) (*model.Inbound, error) {
  1272. if inbound == nil {
  1273. return nil, fmt.Errorf("inbound is nil")
  1274. }
  1275. runtimeInbound := *inbound
  1276. settings := map[string]any{}
  1277. if err := json.Unmarshal([]byte(inbound.Settings), &settings); err != nil {
  1278. return nil, err
  1279. }
  1280. clients, ok := settings["clients"].([]any)
  1281. if !ok {
  1282. return &runtimeInbound, nil
  1283. }
  1284. var clientStats []xray.ClientTraffic
  1285. err := tx.Model(xray.ClientTraffic{}).
  1286. Where("inbound_id = ?", inbound.Id).
  1287. Select("email", "enable").
  1288. Find(&clientStats).Error
  1289. if err != nil {
  1290. return nil, err
  1291. }
  1292. enableMap := make(map[string]bool, len(clientStats))
  1293. for _, clientTraffic := range clientStats {
  1294. enableMap[clientTraffic.Email] = clientTraffic.Enable
  1295. }
  1296. finalClients := make([]any, 0, len(clients))
  1297. for _, client := range clients {
  1298. c, ok := client.(map[string]any)
  1299. if !ok {
  1300. continue
  1301. }
  1302. email, _ := c["email"].(string)
  1303. if enable, exists := enableMap[email]; exists && !enable {
  1304. continue
  1305. }
  1306. if manualEnable, ok := c["enable"].(bool); ok && !manualEnable {
  1307. continue
  1308. }
  1309. finalClients = append(finalClients, c)
  1310. }
  1311. settings["clients"] = finalClients
  1312. modifiedSettings, err := json.MarshalIndent(settings, "", " ")
  1313. if err != nil {
  1314. return nil, err
  1315. }
  1316. runtimeInbound.Settings = string(modifiedSettings)
  1317. return &runtimeInbound, nil
  1318. }
  1319. // updateClientTraffics syncs the ClientTraffic rows with the inbound's clients
  1320. // list: removes rows for emails that disappeared, inserts rows for newly-added
  1321. // emails. Uses sets for O(N) lookup — the previous nested-loop implementation
  1322. // was O(N²) and degraded into multi-second pauses on inbounds with thousands
  1323. // of clients (toggling, saving, or deleting any such inbound felt frozen).
  1324. func (s *InboundService) updateClientTraffics(tx *gorm.DB, oldInbound *model.Inbound, newInbound *model.Inbound) error {
  1325. oldClients, err := s.GetClients(oldInbound)
  1326. if err != nil {
  1327. return err
  1328. }
  1329. newClients, err := s.GetClients(newInbound)
  1330. if err != nil {
  1331. return err
  1332. }
  1333. // Email is the unique key for ClientTraffic rows. Clients without an
  1334. // email have no stats row to sync — skip them on both sides instead of
  1335. // risking a unique-constraint hit or accidental delete of an unrelated row.
  1336. oldEmails := make(map[string]struct{}, len(oldClients))
  1337. for i := range oldClients {
  1338. if oldClients[i].Email == "" {
  1339. continue
  1340. }
  1341. oldEmails[oldClients[i].Email] = struct{}{}
  1342. }
  1343. newEmails := make(map[string]struct{}, len(newClients))
  1344. for i := range newClients {
  1345. if newClients[i].Email == "" {
  1346. continue
  1347. }
  1348. newEmails[newClients[i].Email] = struct{}{}
  1349. }
  1350. // Drop stats rows for removed emails — but not when a sibling inbound
  1351. // still references the email, since the row is the shared accumulator.
  1352. for i := range oldClients {
  1353. email := oldClients[i].Email
  1354. if email == "" {
  1355. continue
  1356. }
  1357. if _, kept := newEmails[email]; kept {
  1358. continue
  1359. }
  1360. stillUsed, err := s.emailUsedByOtherInbounds(email, oldInbound.Id)
  1361. if err != nil {
  1362. return err
  1363. }
  1364. if stillUsed {
  1365. continue
  1366. }
  1367. if err := s.DelClientStat(tx, email); err != nil {
  1368. return err
  1369. }
  1370. // Keep inbound_client_ips in sync when the inbound edit drops an
  1371. // email, so the IP-limit job doesn't keep a ghost tracking row (#4963).
  1372. if err := s.DelClientIPs(tx, email); err != nil {
  1373. return err
  1374. }
  1375. }
  1376. for i := range newClients {
  1377. email := newClients[i].Email
  1378. if email == "" {
  1379. continue
  1380. }
  1381. if _, existed := oldEmails[email]; existed {
  1382. if err := s.UpdateClientStat(tx, email, &newClients[i]); err != nil {
  1383. return err
  1384. }
  1385. continue
  1386. }
  1387. if err := s.AddClientStat(tx, oldInbound.Id, &newClients[i]); err != nil {
  1388. return err
  1389. }
  1390. }
  1391. return nil
  1392. }
  1393. func (s *InboundService) GetInboundTags() (string, error) {
  1394. db := database.GetDB()
  1395. var inboundTags []string
  1396. err := db.Model(model.Inbound{}).Select("tag").Find(&inboundTags).Error
  1397. if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
  1398. return "", err
  1399. }
  1400. tags, _ := json.Marshal(inboundTags)
  1401. return string(tags), nil
  1402. }
  1403. func (s *InboundService) GetClientReverseTags() (string, error) {
  1404. db := database.GetDB()
  1405. var inbounds []model.Inbound
  1406. err := db.Model(model.Inbound{}).Select("settings").Where("protocol = ?", "vless").Find(&inbounds).Error
  1407. if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
  1408. return "[]", err
  1409. }
  1410. tagSet := make(map[string]struct{})
  1411. for _, inbound := range inbounds {
  1412. var settings map[string]any
  1413. if err := json.Unmarshal([]byte(inbound.Settings), &settings); err != nil {
  1414. continue
  1415. }
  1416. clients, ok := settings["clients"].([]any)
  1417. if !ok {
  1418. continue
  1419. }
  1420. for _, client := range clients {
  1421. clientMap, ok := client.(map[string]any)
  1422. if !ok {
  1423. continue
  1424. }
  1425. reverse, ok := clientMap["reverse"].(map[string]any)
  1426. if !ok {
  1427. continue
  1428. }
  1429. tag, _ := reverse["tag"].(string)
  1430. tag = strings.TrimSpace(tag)
  1431. if tag != "" {
  1432. tagSet[tag] = struct{}{}
  1433. }
  1434. }
  1435. }
  1436. rawTags := make([]string, 0, len(tagSet))
  1437. for tag := range tagSet {
  1438. rawTags = append(rawTags, tag)
  1439. }
  1440. sort.Strings(rawTags)
  1441. result, _ := json.Marshal(rawTags)
  1442. return string(result), nil
  1443. }
  1444. func (s *InboundService) SearchInbounds(query string) ([]*model.Inbound, error) {
  1445. db := database.GetDB()
  1446. var inbounds []*model.Inbound
  1447. err := db.Model(model.Inbound{}).Preload("ClientStats").Where("remark like ?", "%"+query+"%").Find(&inbounds).Error
  1448. if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
  1449. return nil, err
  1450. }
  1451. return inbounds, nil
  1452. }