client_inbound_apply.go 46 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476
  1. package service
  2. import (
  3. "context"
  4. "encoding/json"
  5. "fmt"
  6. "maps"
  7. "strings"
  8. "time"
  9. "github.com/mhsanaei/3x-ui/v3/internal/database"
  10. "github.com/mhsanaei/3x-ui/v3/internal/database/model"
  11. "github.com/mhsanaei/3x-ui/v3/internal/logger"
  12. "github.com/mhsanaei/3x-ui/v3/internal/util/common"
  13. "github.com/mhsanaei/3x-ui/v3/internal/util/random"
  14. "github.com/mhsanaei/3x-ui/v3/internal/web/runtime"
  15. "github.com/mhsanaei/3x-ui/v3/internal/xray"
  16. "gorm.io/gorm"
  17. )
  18. // droppedClientNeedsRestart is what restartXrayOnClientDisable asks for: the core
  19. // API drops the credential only, so a live session needs the process replaced.
  20. func droppedClientNeedsRestart() bool {
  21. on, err := (&SettingService{}).GetRestartXrayOnClientDisable()
  22. if err != nil {
  23. logger.Warning("get RestartXrayOnClientDisable failed:", err)
  24. }
  25. return on
  26. }
  27. func sameClientConfigExceptUpdatedAt(a, b map[string]any) bool {
  28. aa := maps.Clone(a)
  29. bb := maps.Clone(b)
  30. delete(aa, "updated_at")
  31. delete(bb, "updated_at")
  32. an, aerr := json.Marshal(aa)
  33. bn, berr := json.Marshal(bb)
  34. return aerr == nil && berr == nil && string(an) == string(bn)
  35. }
  36. // advancePushedInbound advances the node's reconcile-skip fingerprint to what the
  37. // per-client pushes delivered, not the saved settings a traffic tick may have extended.
  38. func advancePushedInbound(rt runtime.Runtime, prevSettings, pushedSettings string, ib *model.Inbound) {
  39. rem, ok := rt.(*runtime.Remote)
  40. if !ok {
  41. return
  42. }
  43. prev := *ib
  44. prev.Settings = prevSettings
  45. pushed := *ib
  46. pushed.Settings = pushedSettings
  47. rem.AdvancePushedInbound(&prev, &pushed)
  48. }
  49. // delInboundClients removes several clients from a single inbound in one pass:
  50. // one settings rewrite, one runtime sweep, one Save and one link delta for the
  51. // whole batch, instead of repeating the full per-client cycle. It mirrors the
  52. // semantics of DelInboundClientByEmail for each removed client. needRestart is
  53. // the OR across all removals.
  54. func (s *ClientService) delInboundClients(inboundSvc *InboundService, inboundId int, recs []*model.ClientRecord, keepTraffic bool) (bool, error) {
  55. if len(recs) == 0 {
  56. return false, nil
  57. }
  58. defer lockInbound(inboundId).Unlock()
  59. oldInbound, err := inboundSvc.GetInbound(inboundId)
  60. if err != nil {
  61. logger.Error("Load Old Data Error")
  62. return false, err
  63. }
  64. var settings map[string]any
  65. if err := json.Unmarshal([]byte(oldInbound.Settings), &settings); err != nil {
  66. return false, err
  67. }
  68. // Match by email — the client's stable identity (see Delete). Removes every
  69. // entry carrying a wanted email, independent of credential drift.
  70. wanted := make(map[string]struct{}, len(recs))
  71. for _, rec := range recs {
  72. if rec.Email != "" {
  73. wanted[rec.Email] = struct{}{}
  74. }
  75. }
  76. interfaceClients, ok := settings["clients"].([]any)
  77. if !ok {
  78. return false, common.NewError("invalid clients format in inbound settings")
  79. }
  80. type removedClient struct {
  81. email string
  82. needApiDel bool
  83. }
  84. removed := make([]removedClient, 0, len(wanted))
  85. newClients := make([]any, 0, len(interfaceClients))
  86. for _, client := range interfaceClients {
  87. c, ok := client.(map[string]any)
  88. if !ok {
  89. newClients = append(newClients, client)
  90. continue
  91. }
  92. email, _ := c["email"].(string)
  93. if _, hit := wanted[email]; hit && email != "" {
  94. enable, _ := c["enable"].(bool)
  95. removed = append(removed, removedClient{email: email, needApiDel: enable})
  96. continue
  97. }
  98. newClients = append(newClients, client)
  99. }
  100. if len(removed) == 0 {
  101. return false, nil
  102. }
  103. db := database.GetDB()
  104. newClients = compactOrphans(db, newClients)
  105. if newClients == nil {
  106. newClients = []any{}
  107. }
  108. settings["clients"] = newClients
  109. newSettings, err := json.MarshalIndent(settings, "", " ")
  110. if err != nil {
  111. return false, err
  112. }
  113. prevSettings := oldInbound.Settings
  114. oldInbound.Settings = string(newSettings)
  115. var sharedSet map[string]bool
  116. if !keepTraffic {
  117. removedEmails := make([]string, 0, len(removed))
  118. for _, r := range removed {
  119. if r.email != "" {
  120. removedEmails = append(removedEmails, r.email)
  121. }
  122. }
  123. var sharedErr error
  124. sharedSet, sharedErr = inboundSvc.emailsUsedByOtherInbounds(removedEmails, inboundId)
  125. if sharedErr != nil {
  126. return false, sharedErr
  127. }
  128. }
  129. needRestart := false
  130. // Read each client's live state before the DB write (DelClientStat would
  131. // erase the enable flag we need to decide on a runtime removal).
  132. type delTarget struct {
  133. email string
  134. emailShared bool
  135. notDepleted bool
  136. needApiDel bool
  137. }
  138. targets := make([]delTarget, 0, len(removed))
  139. for _, r := range removed {
  140. email := r.email
  141. emailShared := sharedSet[strings.ToLower(strings.TrimSpace(email))]
  142. notDepleted := false
  143. if len(email) > 0 {
  144. var enables []bool
  145. if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email).Limit(1).Pluck("enable", &enables).Error; err != nil {
  146. logger.Error("Get stats error")
  147. return needRestart, err
  148. }
  149. notDepleted = len(enables) > 0 && enables[0]
  150. }
  151. targets = append(targets, delTarget{email: email, emailShared: emailShared, notDepleted: notDepleted, needApiDel: r.needApiDel})
  152. }
  153. // Persist the batch deletion atomically, serialized against the traffic poll
  154. // to avoid the cross-transaction lock-order deadlock (runSerializedTx).
  155. if txErr := runSerializedTx(func(tx *gorm.DB) error {
  156. for _, t := range targets {
  157. if t.emailShared || keepTraffic {
  158. continue
  159. }
  160. if e := inboundSvc.DelClientIPs(tx, t.email); e != nil {
  161. logger.Error("Error in delete client IPs")
  162. return e
  163. }
  164. if len(t.email) > 0 {
  165. if e := inboundSvc.DelClientStat(tx, t.email); e != nil {
  166. logger.Error("Delete stats Data Error")
  167. return e
  168. }
  169. }
  170. }
  171. if e := commitInboundClientSettings(tx, oldInbound, prevSettings); e != nil {
  172. return e
  173. }
  174. detached := make([]string, 0, len(targets))
  175. for _, t := range targets {
  176. if t.email != "" {
  177. detached = append(detached, t.email)
  178. }
  179. }
  180. if err := s.ApplyInboundClientDelta(tx, inboundId, nil, detached); err != nil {
  181. return err
  182. }
  183. if oldInbound.NodeID != nil {
  184. return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
  185. }
  186. return nil
  187. }); txErr != nil {
  188. return needRestart, txErr
  189. }
  190. // Resolve the node push plan once for the whole batch instead of per email.
  191. var nodeRt runtime.Runtime
  192. nodePush := false
  193. if oldInbound.NodeID != nil {
  194. rt, push, _, perr := inboundSvc.nodePushPlan(oldInbound)
  195. if perr != nil {
  196. return needRestart, perr
  197. }
  198. nodeRt, nodePush = rt, push
  199. // Large batches collapse into one reconcile push rather than M deletes.
  200. if nodePush && len(targets) > nodeBulkPushThreshold {
  201. nodePush = false
  202. }
  203. }
  204. // Apply runtime deletes after commit — outside the serialized writer so a
  205. // slow node call can't stall traffic accounting.
  206. nodePushFailed := false
  207. for _, t := range targets {
  208. if len(t.email) == 0 {
  209. continue
  210. }
  211. if oldInbound.NodeID == nil {
  212. if t.needApiDel && t.notDepleted {
  213. rt, rterr := inboundSvc.runtimeFor(oldInbound)
  214. if rterr != nil {
  215. needRestart = true
  216. } else if err1 := rt.RemoveUser(context.Background(), oldInbound, t.email); err1 != nil {
  217. if !strings.Contains(err1.Error(), fmt.Sprintf("User %s not found.", t.email)) {
  218. needRestart = true
  219. }
  220. }
  221. }
  222. } else if nodePush && !nodePushFailed {
  223. // First failure ends the batch push; the reconcile converges the rest.
  224. ctx, cancel := nodePushContext()
  225. err1 := nodeRt.DeleteUser(ctx, oldInbound, t.email)
  226. cancel()
  227. if err1 != nil {
  228. logger.Warning("Error in deleting client on", nodeRt.Name(), ":", err1)
  229. nodePushFailed = true
  230. }
  231. }
  232. }
  233. if nodePush && !nodePushFailed {
  234. advancePushedInbound(nodeRt, prevSettings, string(newSettings), oldInbound)
  235. }
  236. return needRestart, nil
  237. }
  238. // otherTunnelAllowedIPs maps every AllowedIPs entry claimed on another
  239. // WireGuard/AmneziaWG inbound to a description of which one holds it: the
  240. // per-inbound defaulters only check their own client list, so two inbounds
  241. // sharing a subnet could otherwise hand out the same address. Disabled
  242. // siblings count too, keeping their addresses reserved for a later re-enable.
  243. //
  244. // selfEmails skips this identity's own entries. Email is globally unique, so a
  245. // match there is never a real collision -- and Attach deliberately reuses one
  246. // address across every inbound it attaches the identity to.
  247. func (s *ClientService) otherTunnelAllowedIPs(db *gorm.DB, inboundSvc *InboundService, excludeID int, selfEmails map[string]struct{}) (map[string]string, error) {
  248. var inbounds []*model.Inbound
  249. err := db.Model(model.Inbound{}).
  250. Where("protocol IN ? AND id != ?", []model.Protocol{model.WireGuard, model.AmneziaWG}, excludeID).
  251. Find(&inbounds).Error
  252. if err != nil {
  253. return nil, err
  254. }
  255. used := make(map[string]string)
  256. for _, ib := range inbounds {
  257. clients, cErr := inboundSvc.GetClients(ib)
  258. if cErr != nil {
  259. continue
  260. }
  261. name := ib.Remark
  262. if name == "" {
  263. name = ib.Tag
  264. }
  265. label := fmt.Sprintf("inbound '%s' (#%d)", name, ib.Id)
  266. for _, c := range clients {
  267. if _, self := selfEmails[strings.ToLower(c.Email)]; self {
  268. continue
  269. }
  270. for _, addr := range c.AllowedIPs {
  271. used[addr] = label
  272. }
  273. }
  274. }
  275. return used, nil
  276. }
  277. func (s *ClientService) checkEmailsExistForClients(inboundSvc *InboundService, clients []model.Client) (string, error) {
  278. emailSubIDs, err := inboundSvc.emailSubIDsForClients(clients)
  279. if err != nil {
  280. return "", err
  281. }
  282. seen := make(map[string]string, len(clients))
  283. for _, client := range clients {
  284. if client.Email == "" {
  285. continue
  286. }
  287. key := strings.ToLower(client.Email)
  288. if prev, ok := seen[key]; ok {
  289. if prev != client.SubID || client.SubID == "" {
  290. return client.Email, nil
  291. }
  292. continue
  293. }
  294. seen[key] = client.SubID
  295. if existingSub, ok := emailSubIDs[key]; ok {
  296. if client.SubID == "" || existingSub == "" || existingSub != client.SubID {
  297. return client.Email, nil
  298. }
  299. }
  300. }
  301. return "", nil
  302. }
  303. func (s *ClientService) AddInboundClient(inboundSvc *InboundService, data *model.Inbound) (bool, error) {
  304. defer lockInbound(data.Id).Unlock()
  305. clients, err := inboundSvc.GetClients(data)
  306. if err != nil {
  307. return false, err
  308. }
  309. if err := validateClientsRenewal(clients); err != nil {
  310. return false, err
  311. }
  312. var settings map[string]any
  313. err = json.Unmarshal([]byte(data.Settings), &settings)
  314. if err != nil {
  315. return false, err
  316. }
  317. interfaceClients := settings["clients"].([]any)
  318. nowTs := time.Now().Unix() * 1000
  319. for i := range interfaceClients {
  320. if cm, ok := interfaceClients[i].(map[string]any); ok {
  321. if _, ok2 := cm["created_at"]; !ok2 {
  322. cm["created_at"] = nowTs
  323. }
  324. cm["updated_at"] = nowTs
  325. existingSub, _ := cm["subId"].(string)
  326. if strings.TrimSpace(existingSub) == "" {
  327. cm["subId"] = random.NumLower(16)
  328. }
  329. interfaceClients[i] = cm
  330. }
  331. }
  332. existEmail, err := s.checkEmailsExistForClients(inboundSvc, clients)
  333. if err != nil {
  334. return false, err
  335. }
  336. if existEmail != "" {
  337. return false, common.NewError("Duplicate email:", existEmail)
  338. }
  339. oldInbound, err := inboundSvc.GetInbound(data.Id)
  340. if err != nil {
  341. return false, err
  342. }
  343. existingClients, err := inboundSvc.GetClients(oldInbound)
  344. if err != nil {
  345. return false, err
  346. }
  347. // A client already on this inbound is skipped instead of appended again:
  348. // checkEmailsExistForClients exempts a matching subId so one identity can
  349. // live on several inbounds, which let retried or raced adds duplicate the
  350. // same email inside a single settings array (#5770). clients and
  351. // interfaceClients are parsed from the same data.Settings array, so they
  352. // stay index-aligned while filtering.
  353. if len(existingClients) > 0 && len(clients) > 0 {
  354. existingEmails := make(map[string]struct{}, len(existingClients))
  355. for _, c := range existingClients {
  356. if c.Email != "" {
  357. existingEmails[strings.ToLower(c.Email)] = struct{}{}
  358. }
  359. }
  360. keptClients := make([]model.Client, 0, len(clients))
  361. keptWire := make([]any, 0, len(interfaceClients))
  362. for i, c := range clients {
  363. if c.Email != "" {
  364. if _, dup := existingEmails[strings.ToLower(c.Email)]; dup {
  365. continue
  366. }
  367. }
  368. keptClients = append(keptClients, c)
  369. if i < len(interfaceClients) {
  370. keptWire = append(keptWire, interfaceClients[i])
  371. }
  372. }
  373. if len(keptClients) == 0 {
  374. return false, nil
  375. }
  376. clients = keptClients
  377. interfaceClients = keptWire
  378. }
  379. var selfEmails map[string]struct{}
  380. if oldInbound.Protocol == model.WireGuard || oldInbound.Protocol == model.AmneziaWG {
  381. selfEmails = make(map[string]struct{}, len(clients))
  382. for _, c := range clients {
  383. if c.Email != "" {
  384. selfEmails[strings.ToLower(c.Email)] = struct{}{}
  385. }
  386. }
  387. crossUsed, cErr := s.otherTunnelAllowedIPs(database.GetDB(), inboundSvc, oldInbound.Id, selfEmails)
  388. if cErr != nil {
  389. return false, cErr
  390. }
  391. if oldInbound.Protocol == model.WireGuard {
  392. if dErr := defaultWireguardClients(oldInbound.Settings, existingClients, clients, interfaceClients, crossUsed); dErr != nil {
  393. return false, dErr
  394. }
  395. }
  396. if oldInbound.Protocol == model.AmneziaWG {
  397. if dErr := defaultAmneziaWGClients(oldInbound.Settings, existingClients, clients, interfaceClients, crossUsed); dErr != nil {
  398. return false, dErr
  399. }
  400. }
  401. }
  402. var portCtx portConflictContext
  403. if oldInbound.Protocol == model.AmneziaWG {
  404. portCtx, err = inboundSvc.loadPortConflictContext(database.GetDB())
  405. if err != nil {
  406. return false, err
  407. }
  408. }
  409. for _, client := range clients {
  410. if strings.TrimSpace(client.Email) == "" {
  411. return false, common.NewError("client email is required")
  412. }
  413. switch oldInbound.Protocol {
  414. case "trojan":
  415. if client.Password == "" {
  416. return false, common.NewError("empty client ID")
  417. }
  418. case "shadowsocks":
  419. if client.Email == "" {
  420. return false, common.NewError("empty client ID")
  421. }
  422. case "hysteria":
  423. if client.Auth == "" {
  424. return false, common.NewError("empty client ID")
  425. }
  426. case "wireguard", "amneziawg":
  427. if client.PublicKey == "" {
  428. return false, common.NewError("wireguard client requires a key")
  429. }
  430. case "mtproto":
  431. if client.Secret == "" {
  432. return false, common.NewError("mtproto client requires a secret")
  433. }
  434. if client.AdTag != "" && !model.ValidMtprotoAdTag(client.AdTag) {
  435. return false, common.NewError("mtproto client ad tag must be 32 hex characters")
  436. }
  437. case "tuic":
  438. if client.ID == "" {
  439. return false, common.NewError("empty client ID")
  440. }
  441. if client.Password == "" {
  442. return false, common.NewError("tuic client requires a password")
  443. }
  444. if client.Email == "" {
  445. return false, common.NewError("empty client email")
  446. }
  447. default:
  448. if client.ID == "" {
  449. return false, common.NewError("empty client ID")
  450. }
  451. }
  452. if oldInbound.Protocol == model.AmneziaWG {
  453. if hit := inboundSvc.checkForwardedPortsConflict(portCtx, client.ForwardedPorts); hit != "" {
  454. return false, common.NewError("amneziawg: forwardedPorts collides with", hit)
  455. }
  456. }
  457. }
  458. var oldSettings map[string]any
  459. err = json.Unmarshal([]byte(oldInbound.Settings), &oldSettings)
  460. if err != nil {
  461. return false, err
  462. }
  463. if oldInbound.Protocol == model.Shadowsocks {
  464. applyShadowsocksClientMethod(interfaceClients, oldSettings)
  465. }
  466. oldClients, _ := oldSettings["clients"].([]any)
  467. oldClients = compactOrphans(database.GetDB(), oldClients)
  468. oldClients = append(oldClients, interfaceClients...)
  469. oldSettings["clients"] = oldClients
  470. newSettings, err := json.MarshalIndent(oldSettings, "", " ")
  471. if err != nil {
  472. return false, err
  473. }
  474. prevSettings := oldInbound.Settings
  475. oldInbound.Settings = string(newSettings)
  476. // From the stamped wire entries, not from clients: created_at / updated_at /
  477. // subId are written onto interfaceClients above, after clients was parsed.
  478. addedClients, err := settingsEntriesToClients(interfaceClients)
  479. if err != nil {
  480. return false, err
  481. }
  482. needRestart := false
  483. rt, push, _, perr := inboundSvc.nodePushPlan(oldInbound)
  484. if perr != nil {
  485. return false, perr
  486. }
  487. // Persist client stats + inbound atomically, serialized against the traffic
  488. // poll to avoid the cross-transaction lock-order deadlock (runSerializedTx).
  489. if txErr := runSerializedTx(func(tx *gorm.DB) error {
  490. // lockInbound is per-inbound, so the pre-tx cross-inbound checks race
  491. // concurrent writers on other inbounds — re-run them in here (#6225).
  492. if oldInbound.Protocol == model.WireGuard || oldInbound.Protocol == model.AmneziaWG {
  493. crossUsed, cErr := s.otherTunnelAllowedIPs(tx, inboundSvc, oldInbound.Id, selfEmails)
  494. if cErr != nil {
  495. return cErr
  496. }
  497. crossAddrs := make([]string, 0, len(crossUsed))
  498. for addr := range crossUsed {
  499. crossAddrs = append(crossAddrs, addr)
  500. }
  501. for i := range clients {
  502. if entry, taken := wireguardAllowedIPsOverlap(clients[i].AllowedIPs, crossAddrs); taken != "" {
  503. return common.NewError("allowedIPs entry", entry, "overlaps", taken, "used by a client on", crossUsed[taken])
  504. }
  505. }
  506. }
  507. if oldInbound.Protocol == model.AmneziaWG {
  508. txPortCtx, pErr := inboundSvc.loadPortConflictContext(tx)
  509. if pErr != nil {
  510. return pErr
  511. }
  512. for i := range clients {
  513. if hit := inboundSvc.checkForwardedPortsConflict(txPortCtx, clients[i].ForwardedPorts); hit != "" {
  514. return common.NewError("amneziawg: forwardedPorts collides with", hit)
  515. }
  516. }
  517. }
  518. for i := range clients {
  519. if len(clients[i].Email) == 0 {
  520. continue
  521. }
  522. if e := inboundSvc.AddClientStat(tx, data.Id, &clients[i]); e != nil {
  523. return e
  524. }
  525. }
  526. if e := commitInboundClientSettings(tx, oldInbound, prevSettings); e != nil {
  527. return e
  528. }
  529. if err := s.ApplyInboundClientDelta(tx, oldInbound.Id, addedClients, nil); err != nil {
  530. return err
  531. }
  532. if oldInbound.NodeID != nil {
  533. return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
  534. }
  535. return nil
  536. }); txErr != nil {
  537. return false, txErr
  538. }
  539. // Apply to the running runtime after commit — outside the serialized writer
  540. // so a slow node call can't stall traffic accounting.
  541. if oldInbound.NodeID == nil {
  542. if !push {
  543. needRestart = true
  544. } else if oldInbound.Protocol == model.MTProto {
  545. inboundSvc.applyLocalMtproto(oldInbound.Id)
  546. } else if oldInbound.Protocol == model.AmneziaWG {
  547. inboundSvc.applyLocalAmneziaWG(oldInbound.Id)
  548. } else if oldInbound.Protocol == model.TUIC {
  549. inboundSvc.applyLocalTuic(oldInbound.Id)
  550. } else {
  551. for _, client := range clients {
  552. if len(client.Email) == 0 {
  553. needRestart = true
  554. continue
  555. }
  556. if !client.Enable {
  557. continue
  558. }
  559. cipher := ""
  560. if oldInbound.Protocol == "shadowsocks" {
  561. cipher, _ = oldSettings["method"].(string)
  562. }
  563. err1 := rt.AddUser(context.Background(), oldInbound, map[string]any{
  564. "email": client.Email,
  565. "id": client.ID,
  566. "auth": client.Auth,
  567. "security": client.Security,
  568. "flow": client.Flow,
  569. "password": client.Password,
  570. "cipher": cipher,
  571. "publicKey": client.PublicKey,
  572. "allowedIPs": client.AllowedIPs,
  573. "preSharedKey": client.PreSharedKey,
  574. "keepAlive": keepAliveStr(client.KeepAliveSeconds()),
  575. "reverse": client.Reverse,
  576. })
  577. if err1 == nil {
  578. logger.Debug("Client added on", rt.Name(), ":", client.Email)
  579. } else {
  580. logger.Debug("Error in adding client on", rt.Name(), ":", err1)
  581. needRestart = true
  582. }
  583. }
  584. }
  585. } else {
  586. // Large batches would be M sequential per-client RPCs; the inbound's saved
  587. // settings already hold the final set, so mark dirty and let one reconcile
  588. // push converge the node instead.
  589. if push && len(clients) > nodeBulkPushThreshold {
  590. push = false
  591. }
  592. for _, client := range clients {
  593. // /clients/add on the node historically coerced enable=true; skip live
  594. // push for disabled clients and leave dirty so reconcile converges.
  595. if !client.Enable {
  596. push = false
  597. continue
  598. }
  599. if push {
  600. ctx, cancel := nodePushContext()
  601. err1 := rt.AddClient(ctx, oldInbound, client)
  602. cancel()
  603. if err1 != nil {
  604. logger.Warning("Error in adding client on", rt.Name(), ":", err1)
  605. push = false
  606. }
  607. }
  608. }
  609. if push {
  610. advancePushedInbound(rt, prevSettings, string(newSettings), oldInbound)
  611. }
  612. }
  613. return needRestart, nil
  614. }
  615. func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *model.Inbound, oldEmail string) (bool, error) {
  616. defer lockInbound(data.Id).Unlock()
  617. clients, err := inboundSvc.GetClients(data)
  618. if err != nil {
  619. return false, err
  620. }
  621. if err := validateClientsRenewal(clients); err != nil {
  622. return false, err
  623. }
  624. var settings map[string]any
  625. err = json.Unmarshal([]byte(data.Settings), &settings)
  626. if err != nil {
  627. return false, err
  628. }
  629. interfaceClients := settings["clients"].([]any)
  630. oldInbound, err := inboundSvc.GetInbound(data.Id)
  631. if err != nil {
  632. return false, err
  633. }
  634. oldClients, err := inboundSvc.GetClients(oldInbound)
  635. if err != nil {
  636. return false, err
  637. }
  638. newClientId := ""
  639. switch oldInbound.Protocol {
  640. case "trojan":
  641. newClientId = clients[0].Password
  642. case "shadowsocks":
  643. newClientId = clients[0].Email
  644. case "hysteria":
  645. newClientId = clients[0].Auth
  646. case "wireguard", "amneziawg":
  647. newClientId = clients[0].Email
  648. case "mtproto":
  649. newClientId = clients[0].Email
  650. default:
  651. newClientId = clients[0].ID
  652. }
  653. // Locate the client to replace by email — the client's stable identity.
  654. // Credentials (uuid/password/auth) can drift from the inbound JSON, so they
  655. // are never used for matching.
  656. clientIndex := -1
  657. for index, oldClient := range oldClients {
  658. if strings.EqualFold(oldClient.Email, oldEmail) {
  659. oldEmail = oldClient.Email
  660. clientIndex = index
  661. break
  662. }
  663. }
  664. if newClientId == "" || clientIndex == -1 {
  665. return false, common.NewError("empty client ID")
  666. }
  667. if strings.TrimSpace(clients[0].Email) == "" {
  668. return false, common.NewError("client email is required")
  669. }
  670. if oldInbound.Protocol == model.MTProto && clients[0].AdTag != "" && !model.ValidMtprotoAdTag(clients[0].AdTag) {
  671. return false, common.NewError("mtproto client ad tag must be 32 hex characters")
  672. }
  673. if clients[0].Email != oldEmail {
  674. existEmail, err := s.checkEmailsExistForClients(inboundSvc, clients)
  675. if err != nil {
  676. return false, err
  677. }
  678. if existEmail != "" {
  679. return false, common.NewError("Duplicate email:", existEmail)
  680. }
  681. }
  682. // WireGuard/AmneziaWG keys are never rotated by an edit: when the incoming
  683. // payload omits them (a metadata-only change), carry the stored credentials
  684. // forward so the settings JSON and the running peer keep the client's identity.
  685. if (oldInbound.Protocol == model.WireGuard || oldInbound.Protocol == model.AmneziaWG) && clientIndex >= 0 && clientIndex < len(oldClients) {
  686. old := oldClients[clientIndex]
  687. if clients[0].PrivateKey == "" {
  688. clients[0].PrivateKey = old.PrivateKey
  689. }
  690. if clients[0].PublicKey == "" {
  691. clients[0].PublicKey = old.PublicKey
  692. }
  693. if len(clients[0].AllowedIPs) == 0 {
  694. clients[0].AllowedIPs = old.AllowedIPs
  695. } else {
  696. normalized, nErr := normalizeWireguardAllowedIPs(clients[0].AllowedIPs)
  697. if nErr != nil {
  698. return false, nErr
  699. }
  700. if len(normalized) == 0 {
  701. clients[0].AllowedIPs = old.AllowedIPs
  702. } else {
  703. peers := make([]string, 0, len(oldClients))
  704. for i := range oldClients {
  705. if i == clientIndex {
  706. continue
  707. }
  708. peers = append(peers, oldClients[i].AllowedIPs...)
  709. }
  710. if entry, taken := wireguardAllowedIPsOverlap(normalized, peers); taken != "" {
  711. return false, common.NewError("wireguard: allowedIPs entry", entry, "overlaps", taken, "used by another client")
  712. }
  713. clients[0].AllowedIPs = normalized
  714. }
  715. }
  716. if clients[0].PreSharedKey == "" {
  717. clients[0].PreSharedKey = old.PreSharedKey
  718. }
  719. if clients[0].KeepAlive == nil {
  720. clients[0].KeepAlive = old.KeepAlive
  721. }
  722. // ForwardedPorts is AmneziaWG-only (WireGuard's own inbound never
  723. // reads it), same carry-forward reasoning as the fields above: a
  724. // partial edit (e.g. a Telegram-bot enable/expiry toggle, or an API
  725. // call that omits the field) must not silently drop a client's
  726. // existing port-forwarding spec.
  727. if oldInbound.Protocol == model.AmneziaWG && clients[0].ForwardedPorts == "" {
  728. clients[0].ForwardedPorts = old.ForwardedPorts
  729. }
  730. }
  731. if oldInbound.Protocol == model.AmneziaWG {
  732. portCtx, err := inboundSvc.loadPortConflictContext(database.GetDB())
  733. if err != nil {
  734. return false, err
  735. }
  736. if hit := inboundSvc.checkForwardedPortsConflict(portCtx, clients[0].ForwardedPorts); hit != "" {
  737. return false, common.NewError("amneziawg: forwardedPorts collides with", hit)
  738. }
  739. }
  740. var oldSettings map[string]any
  741. err = json.Unmarshal([]byte(oldInbound.Settings), &oldSettings)
  742. if err != nil {
  743. return false, err
  744. }
  745. settingsClients, _ := oldSettings["clients"].([]any)
  746. var preservedCreated any
  747. var preservedSubID string
  748. var oldClientMap map[string]any
  749. if clientIndex >= 0 && clientIndex < len(settingsClients) {
  750. if oldMap, ok := settingsClients[clientIndex].(map[string]any); ok {
  751. oldClientMap = oldMap
  752. if v, ok2 := oldMap["created_at"]; ok2 {
  753. preservedCreated = v
  754. }
  755. preservedSubID, _ = oldMap["subId"].(string)
  756. }
  757. }
  758. if oldInbound.Protocol == model.Shadowsocks {
  759. applyShadowsocksClientMethod(interfaceClients, oldSettings)
  760. }
  761. if len(interfaceClients) > 0 {
  762. if newMap, ok := interfaceClients[0].(map[string]any); ok {
  763. if preservedCreated == nil {
  764. preservedCreated = time.Now().Unix() * 1000
  765. }
  766. newMap["created_at"] = preservedCreated
  767. newSub, _ := newMap["subId"].(string)
  768. if strings.TrimSpace(newSub) == "" {
  769. if strings.TrimSpace(preservedSubID) != "" {
  770. newMap["subId"] = preservedSubID
  771. } else {
  772. newMap["subId"] = random.NumLower(16)
  773. }
  774. }
  775. if v, ok2 := newMap["subId"].(string); ok2 {
  776. clients[0].SubID = v
  777. }
  778. if oldInbound.Protocol == model.WireGuard || oldInbound.Protocol == model.AmneziaWG {
  779. newMap["privateKey"] = clients[0].PrivateKey
  780. newMap["publicKey"] = clients[0].PublicKey
  781. newMap["allowedIPs"] = clients[0].AllowedIPs
  782. if clients[0].PreSharedKey != "" {
  783. newMap["preSharedKey"] = clients[0].PreSharedKey
  784. }
  785. if ka := clients[0].KeepAliveSeconds(); ka > 0 {
  786. newMap["keepAlive"] = ka
  787. }
  788. if oldInbound.Protocol == model.AmneziaWG && clients[0].ForwardedPorts != "" {
  789. newMap["forwardedPorts"] = clients[0].ForwardedPorts
  790. }
  791. }
  792. if oldClientMap != nil && sameClientConfigExceptUpdatedAt(oldClientMap, newMap) {
  793. if v, ok2 := oldClientMap["updated_at"]; ok2 {
  794. newMap["updated_at"] = v
  795. } else {
  796. delete(newMap, "updated_at")
  797. }
  798. } else {
  799. newMap["updated_at"] = time.Now().Unix() * 1000
  800. }
  801. interfaceClients[0] = newMap
  802. }
  803. }
  804. settingsClients[clientIndex] = interfaceClients[0]
  805. oldSettings["clients"] = settingsClients
  806. if oldInbound.Protocol == model.VLESS {
  807. hasVisionFlow := false
  808. for _, c := range settingsClients {
  809. cm, ok := c.(map[string]any)
  810. if !ok {
  811. continue
  812. }
  813. if flow, _ := cm["flow"].(string); flow == "xtls-rprx-vision" {
  814. hasVisionFlow = true
  815. break
  816. }
  817. }
  818. if !hasVisionFlow {
  819. delete(oldSettings, "testseed")
  820. }
  821. }
  822. newSettings, err := json.MarshalIndent(oldSettings, "", " ")
  823. if err != nil {
  824. return false, err
  825. }
  826. if string(newSettings) == oldInbound.Settings {
  827. return false, nil
  828. }
  829. prevSettings := oldInbound.Settings
  830. oldInbound.Settings = string(newSettings)
  831. // From the stamped wire entry, not from clients[0]: created_at, the
  832. // preserved subId and the WireGuard carry-forward land on interfaceClients.
  833. changedClients, err := settingsEntriesToClients(interfaceClients[:1])
  834. if err != nil {
  835. return false, err
  836. }
  837. var detachEmails []string
  838. if len(oldEmail) > 0 && oldEmail != clients[0].Email {
  839. detachEmails = []string{oldEmail}
  840. }
  841. needRestart := false
  842. // Resolve the push plan before the DB write so a node-state lookup failure
  843. // still aborts the whole update without committing anything (it used to roll
  844. // the transaction back). nodePushPlan only reads, so order doesn't matter.
  845. var rt runtime.Runtime
  846. var push bool
  847. if len(oldEmail) > 0 {
  848. var perr error
  849. rt, push, _, perr = inboundSvc.nodePushPlan(oldInbound)
  850. if perr != nil {
  851. return false, perr
  852. }
  853. }
  854. // Persist client stats + inbound atomically, serialized against the traffic
  855. // poll to avoid the cross-transaction lock-order deadlock (runSerializedTx).
  856. if txErr := runSerializedTx(func(tx *gorm.DB) error {
  857. // Same re-check-inside-the-writer rule as AddInboundClient (#6225):
  858. // the pre-tx pass can race a concurrent writer on another inbound.
  859. if oldInbound.Protocol == model.AmneziaWG {
  860. txPortCtx, pErr := inboundSvc.loadPortConflictContext(tx)
  861. if pErr != nil {
  862. return pErr
  863. }
  864. if hit := inboundSvc.checkForwardedPortsConflict(txPortCtx, clients[0].ForwardedPorts); hit != "" {
  865. return common.NewError("amneziawg: forwardedPorts collides with", hit)
  866. }
  867. }
  868. if len(clients[0].Email) > 0 {
  869. if len(oldEmail) > 0 {
  870. emailUnchanged := strings.EqualFold(oldEmail, clients[0].Email)
  871. targetExists := int64(0)
  872. if !emailUnchanged {
  873. if e := tx.Model(xray.ClientTraffic{}).Where("email = ?", clients[0].Email).Count(&targetExists).Error; e != nil {
  874. return e
  875. }
  876. }
  877. if emailUnchanged || targetExists == 0 {
  878. if e := inboundSvc.UpdateClientStat(tx, oldEmail, &clients[0]); e != nil {
  879. return e
  880. }
  881. if e := inboundSvc.UpdateClientIPs(tx, oldEmail, clients[0].Email); e != nil {
  882. return e
  883. }
  884. } else {
  885. stillUsed, sErr := inboundSvc.emailUsedByOtherInbounds(oldEmail, data.Id)
  886. if sErr != nil {
  887. return sErr
  888. }
  889. if !stillUsed {
  890. if e := inboundSvc.DelClientStat(tx, oldEmail); e != nil {
  891. return e
  892. }
  893. if e := inboundSvc.DelClientIPs(tx, oldEmail); e != nil {
  894. return e
  895. }
  896. }
  897. if e := inboundSvc.UpdateClientStat(tx, clients[0].Email, &clients[0]); e != nil {
  898. return e
  899. }
  900. }
  901. } else {
  902. if e := inboundSvc.AddClientStat(tx, data.Id, &clients[0]); e != nil {
  903. return e
  904. }
  905. }
  906. } else {
  907. stillUsed, sErr := inboundSvc.emailUsedByOtherInbounds(oldEmail, data.Id)
  908. if sErr != nil {
  909. return sErr
  910. }
  911. if !stillUsed {
  912. if e := inboundSvc.DelClientStat(tx, oldEmail); e != nil {
  913. return e
  914. }
  915. if e := inboundSvc.DelClientIPs(tx, oldEmail); e != nil {
  916. return e
  917. }
  918. }
  919. }
  920. if e := commitInboundClientSettings(tx, oldInbound, prevSettings); e != nil {
  921. return e
  922. }
  923. // Rename the client record in the same transaction as the settings JSON
  924. // so no concurrent SyncInbound can see one renamed without the other.
  925. // Byte-level compare (not EqualFold): case-only edits must rename too,
  926. // otherwise SyncInbound's case-sensitive lookup creates a duplicate row.
  927. if len(oldEmail) > 0 && oldEmail != clients[0].Email {
  928. var renameTaken int64
  929. if e := tx.Model(&model.ClientRecord{}).Where("email = ?", clients[0].Email).Count(&renameTaken).Error; e != nil {
  930. return e
  931. }
  932. if renameTaken == 0 {
  933. if e := tx.Model(&model.ClientRecord{}).Where("email = ?", oldEmail).Update("email", clients[0].Email).Error; e != nil {
  934. return e
  935. }
  936. }
  937. }
  938. // detachEmails covers the rename the guard above refused: the old record
  939. // keeps this inbound's link otherwise, which the full sync used to drop.
  940. if err := s.ApplyInboundClientDelta(tx, oldInbound.Id, changedClients, detachEmails); err != nil {
  941. return err
  942. }
  943. if oldInbound.NodeID != nil {
  944. return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
  945. }
  946. return nil
  947. }); txErr != nil {
  948. return false, txErr
  949. }
  950. // Apply to the running runtime after the DB is committed — outside the
  951. // serialized writer so a slow node call can't stall traffic accounting.
  952. if len(oldEmail) > 0 {
  953. if oldInbound.NodeID == nil {
  954. if !push {
  955. needRestart = true
  956. } else if oldInbound.Protocol == model.MTProto {
  957. inboundSvc.applyLocalMtproto(oldInbound.Id)
  958. } else if oldInbound.Protocol == model.AmneziaWG {
  959. inboundSvc.applyLocalAmneziaWG(oldInbound.Id)
  960. } else if oldInbound.Protocol == model.TUIC {
  961. inboundSvc.applyLocalTuic(oldInbound.Id)
  962. } else {
  963. if oldClients[clientIndex].Enable {
  964. err1 := rt.RemoveUser(context.Background(), oldInbound, oldEmail)
  965. if err1 == nil {
  966. logger.Debug("Old client deleted on", rt.Name(), ":", oldEmail)
  967. // The API removal is enough only while the client is re-added; a
  968. // dropped one ends its session only through a restart.
  969. if !clients[0].Enable && droppedClientNeedsRestart() {
  970. needRestart = true
  971. }
  972. } else if strings.Contains(err1.Error(), fmt.Sprintf("User %s not found.", oldEmail)) {
  973. logger.Debug("User is already deleted. Nothing to do more...")
  974. } else {
  975. logger.Debug("Error in deleting client on", rt.Name(), ":", err1)
  976. needRestart = true
  977. }
  978. }
  979. if clients[0].Enable {
  980. cipher := ""
  981. if oldInbound.Protocol == "shadowsocks" {
  982. cipher, _ = oldSettings["method"].(string)
  983. }
  984. err1 := rt.AddUser(context.Background(), oldInbound, map[string]any{
  985. "email": clients[0].Email,
  986. "id": clients[0].ID,
  987. "security": clients[0].Security,
  988. "flow": clients[0].Flow,
  989. "auth": clients[0].Auth,
  990. "password": clients[0].Password,
  991. "cipher": cipher,
  992. "publicKey": clients[0].PublicKey,
  993. "allowedIPs": clients[0].AllowedIPs,
  994. "preSharedKey": clients[0].PreSharedKey,
  995. "keepAlive": keepAliveStr(clients[0].KeepAliveSeconds()),
  996. "reverse": clients[0].Reverse,
  997. })
  998. if err1 == nil {
  999. logger.Debug("Client edited on", rt.Name(), ":", clients[0].Email)
  1000. } else {
  1001. logger.Debug("Error in adding client on", rt.Name(), ":", err1)
  1002. needRestart = true
  1003. }
  1004. }
  1005. }
  1006. } else if push {
  1007. ctx, cancel := nodePushContext()
  1008. err1 := rt.UpdateUser(ctx, oldInbound, oldEmail, clients[0])
  1009. cancel()
  1010. if err1 != nil {
  1011. logger.Warning("Error in updating client on", rt.Name(), ":", err1)
  1012. } else {
  1013. advancePushedInbound(rt, prevSettings, string(newSettings), oldInbound)
  1014. }
  1015. }
  1016. } else {
  1017. logger.Debug("Client old email not found")
  1018. needRestart = true
  1019. }
  1020. return needRestart, nil
  1021. }
  1022. func (s *ClientService) DelInboundClientByEmail(inboundSvc *InboundService, inboundId int, email string, keepTraffic bool, fullDelete bool) (bool, error) {
  1023. defer lockInbound(inboundId).Unlock()
  1024. oldInbound, err := inboundSvc.GetInbound(inboundId)
  1025. if err != nil {
  1026. logger.Error("Load Old Data Error")
  1027. return false, err
  1028. }
  1029. var settings map[string]any
  1030. if err := json.Unmarshal([]byte(oldInbound.Settings), &settings); err != nil {
  1031. return false, err
  1032. }
  1033. interfaceClients, ok := settings["clients"].([]any)
  1034. if !ok {
  1035. return false, common.NewError("invalid clients format in inbound settings")
  1036. }
  1037. var newClients []any
  1038. needApiDel := false
  1039. found := false
  1040. for _, client := range interfaceClients {
  1041. c, ok := client.(map[string]any)
  1042. if !ok {
  1043. continue
  1044. }
  1045. if cEmail, ok := c["email"].(string); ok && cEmail == email {
  1046. found = true
  1047. needApiDel, _ = c["enable"].(bool)
  1048. } else {
  1049. newClients = append(newClients, client)
  1050. }
  1051. }
  1052. if !found {
  1053. return false, fmt.Errorf("%w for email: %s", ErrClientNotInInbound, email)
  1054. }
  1055. db := database.GetDB()
  1056. newClients = compactOrphans(db, newClients)
  1057. if newClients == nil {
  1058. newClients = []any{}
  1059. }
  1060. settings["clients"] = newClients
  1061. newSettings, err := json.MarshalIndent(settings, "", " ")
  1062. if err != nil {
  1063. return false, err
  1064. }
  1065. prevSettings := oldInbound.Settings
  1066. oldInbound.Settings = string(newSettings)
  1067. emailShared, err := inboundSvc.emailUsedByOtherInbounds(email, inboundId)
  1068. if err != nil {
  1069. return false, err
  1070. }
  1071. needRestart := false
  1072. // Decide what to delete and the push plan before the serialized DB write —
  1073. // these are reads, and nodePushPlan failing should abort before committing.
  1074. delStat := false
  1075. if len(email) > 0 && !emailShared && !keepTraffic {
  1076. traffic, tErr := inboundSvc.GetClientTrafficByEmail(email)
  1077. if tErr != nil {
  1078. return false, tErr
  1079. }
  1080. delStat = traffic != nil
  1081. }
  1082. // The runtime user is scoped to this inbound's tag + email, so the push plan
  1083. // is resolved independently of emailShared — a sibling inbound still carrying
  1084. // the email must not suppress removing the user from this inbound's Xray.
  1085. var rt runtime.Runtime
  1086. var push bool
  1087. if len(email) > 0 && (oldInbound.NodeID != nil || needApiDel) {
  1088. r, p, _, perr := inboundSvc.nodePushPlan(oldInbound)
  1089. if perr != nil {
  1090. return false, perr
  1091. }
  1092. rt, push = r, p
  1093. }
  1094. // Persist the deletion atomically, serialized against the traffic poll to
  1095. // avoid the cross-transaction lock-order deadlock (runSerializedTx).
  1096. if txErr := runSerializedTx(func(tx *gorm.DB) error {
  1097. if !emailShared && !keepTraffic {
  1098. if e := inboundSvc.DelClientIPs(tx, email); e != nil {
  1099. logger.Error("Error in delete client IPs")
  1100. return e
  1101. }
  1102. }
  1103. if delStat {
  1104. if e := inboundSvc.DelClientStat(tx, email); e != nil {
  1105. logger.Error("Delete stats Data Error")
  1106. return e
  1107. }
  1108. }
  1109. if e := commitInboundClientSettings(tx, oldInbound, prevSettings); e != nil {
  1110. return e
  1111. }
  1112. if err := s.ApplyInboundClientDelta(tx, inboundId, nil, []string{email}); err != nil {
  1113. return err
  1114. }
  1115. if oldInbound.NodeID != nil {
  1116. return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
  1117. }
  1118. return nil
  1119. }); txErr != nil {
  1120. return false, txErr
  1121. }
  1122. // Apply the runtime delete after commit — outside the serialized writer so a
  1123. // slow node call can't stall traffic accounting. Independent of emailShared:
  1124. // Xray users are keyed by inbound tag, so the user must be removed from this
  1125. // inbound's runtime even when the same email survives in another inbound.
  1126. if len(email) > 0 {
  1127. if oldInbound.NodeID == nil {
  1128. if oldInbound.Protocol == model.MTProto {
  1129. // mtg serves the full secret set, so any client delete re-applies
  1130. // it (removing the last client stops the sidecar) regardless of the
  1131. // client's enable state.
  1132. inboundSvc.applyLocalMtproto(oldInbound.Id)
  1133. } else if oldInbound.Protocol == model.AmneziaWG {
  1134. // Same reasoning as MTProto above: the interface config is
  1135. // regenerated from the full peer set, so any delete re-applies it.
  1136. inboundSvc.applyLocalAmneziaWG(oldInbound.Id)
  1137. } else if oldInbound.Protocol == model.TUIC {
  1138. inboundSvc.applyLocalTuic(oldInbound.Id)
  1139. } else if needApiDel {
  1140. // Local inbound: a disabled client isn't in the running Xray, so only
  1141. // a live one (needApiDel) needs an API removal.
  1142. if !push {
  1143. needRestart = true
  1144. } else if err1 := rt.RemoveUser(context.Background(), oldInbound, email); err1 == nil {
  1145. logger.Debug("Client deleted on", rt.Name(), ":", email)
  1146. needRestart = droppedClientNeedsRestart()
  1147. } else if strings.Contains(err1.Error(), fmt.Sprintf("User %s not found.", email)) {
  1148. logger.Debug("User is already deleted. Nothing to do more...")
  1149. } else {
  1150. logger.Debug("Error in deleting client on", rt.Name(), ":", email)
  1151. needRestart = true
  1152. }
  1153. }
  1154. } else {
  1155. // Node inbound: propagate the delete regardless of the enable flag —
  1156. // the node's own DB still carries a disabled client and would
  1157. // resurrect it on the next snapshot otherwise. A full client delete
  1158. // must remove the node's client record too, not just detach it from
  1159. // this inbound (#5797).
  1160. if push {
  1161. ctx, cancel := nodePushContext()
  1162. var err1 error
  1163. if fullDelete {
  1164. err1 = rt.DeleteClient(ctx, email)
  1165. } else {
  1166. err1 = rt.DeleteUser(ctx, oldInbound, email)
  1167. }
  1168. cancel()
  1169. if err1 != nil {
  1170. logger.Warning("Error in deleting client on", rt.Name(), ":", err1)
  1171. } else {
  1172. advancePushedInbound(rt, prevSettings, string(newSettings), oldInbound)
  1173. }
  1174. }
  1175. }
  1176. }
  1177. return needRestart, nil
  1178. }
  1179. func (s *ClientService) SetClientTelegramUserID(inboundSvc *InboundService, trafficId int, tgId int64) (bool, error) {
  1180. traffic, inbound, err := inboundSvc.GetClientInboundByTrafficID(trafficId)
  1181. if err != nil {
  1182. return false, err
  1183. }
  1184. if inbound == nil {
  1185. return false, common.NewError("Inbound Not Found For Traffic ID:", trafficId)
  1186. }
  1187. clientEmail := traffic.Email
  1188. oldClients, err := inboundSvc.GetClients(inbound)
  1189. if err != nil {
  1190. return false, err
  1191. }
  1192. found := false
  1193. for _, oldClient := range oldClients {
  1194. if oldClient.Email == clientEmail {
  1195. found = true
  1196. break
  1197. }
  1198. }
  1199. if !found {
  1200. return false, common.NewError("Client Not Found For Email:", clientEmail)
  1201. }
  1202. var settings map[string]any
  1203. err = json.Unmarshal([]byte(inbound.Settings), &settings)
  1204. if err != nil {
  1205. return false, err
  1206. }
  1207. clients := settings["clients"].([]any)
  1208. var newClients []any
  1209. for client_index := range clients {
  1210. c := clients[client_index].(map[string]any)
  1211. if c["email"] == clientEmail {
  1212. c["tgId"] = tgId
  1213. c["updated_at"] = time.Now().Unix() * 1000
  1214. newClients = append(newClients, any(c))
  1215. }
  1216. }
  1217. settings["clients"] = newClients
  1218. modifiedSettings, err := json.MarshalIndent(settings, "", " ")
  1219. if err != nil {
  1220. return false, err
  1221. }
  1222. inbound.Settings = string(modifiedSettings)
  1223. needRestart, err := s.UpdateInboundClient(inboundSvc, inbound, clientEmail)
  1224. return needRestart, err
  1225. }
  1226. func (s *ClientService) CheckIsEnabledByEmail(inboundSvc *InboundService, clientEmail string) (bool, error) {
  1227. _, inbound, err := inboundSvc.GetClientInboundByEmail(clientEmail)
  1228. if err != nil {
  1229. return false, err
  1230. }
  1231. if inbound == nil {
  1232. return false, common.NewError("Inbound Not Found For Email:", clientEmail)
  1233. }
  1234. clients, err := inboundSvc.GetClients(inbound)
  1235. if err != nil {
  1236. return false, err
  1237. }
  1238. isEnable := false
  1239. for _, client := range clients {
  1240. if client.Email == clientEmail {
  1241. isEnable = client.Enable
  1242. break
  1243. }
  1244. }
  1245. return isEnable, err
  1246. }
  1247. func (s *ClientService) ToggleClientEnableByEmail(inboundSvc *InboundService, clientEmail string) (bool, bool, error) {
  1248. current, err := s.CheckIsEnabledByEmail(inboundSvc, clientEmail)
  1249. if err != nil {
  1250. return false, false, err
  1251. }
  1252. target := !current
  1253. needRestart, err := s.applyClientFieldByEmail(inboundSvc, clientEmail, func(c map[string]any) {
  1254. c["enable"] = target
  1255. })
  1256. if err != nil {
  1257. return false, needRestart, err
  1258. }
  1259. return target, needRestart, nil
  1260. }
  1261. func (s *ClientService) SetClientEnableByEmail(inboundSvc *InboundService, clientEmail string, enable bool) (bool, bool, error) {
  1262. current, err := s.CheckIsEnabledByEmail(inboundSvc, clientEmail)
  1263. if err != nil {
  1264. return false, false, err
  1265. }
  1266. if current == enable {
  1267. return false, false, nil
  1268. }
  1269. needRestart, err := s.applyClientFieldByEmail(inboundSvc, clientEmail, func(c map[string]any) {
  1270. c["enable"] = enable
  1271. })
  1272. if err != nil {
  1273. return false, needRestart, err
  1274. }
  1275. return true, needRestart, nil
  1276. }
  1277. // applyClientFieldByEmail loads the inbound currently hosting clientEmail,
  1278. // confirms the client exists, applies mutate to the matching client (plus a
  1279. // refreshed updated_at), and hands a single-client update payload to
  1280. // UpdateInboundClient. The rebuilt clients array intentionally contains only
  1281. // the matched client — that is the input contract UpdateInboundClient expects
  1282. // (clients[0] is the new data; clientEmail locates the row to replace). It
  1283. // backs the single-field by-email setters below.
  1284. // applyClientFieldByEmail mutates a client field on every inbound the email is
  1285. // attached to. A multi-inbound client is one logical identity: patching only
  1286. // the first inbound's JSON would leave the siblings stale, and the next
  1287. // SyncInbound over a stale sibling would revert the edit in the normalized
  1288. // records (#5039).
  1289. func (s *ClientService) applyClientFieldByEmail(inboundSvc *InboundService, clientEmail string, mutate func(c map[string]any)) (bool, error) {
  1290. inboundIds, err := s.GetInboundIdsForEmail(database.GetDB(), clientEmail)
  1291. if err != nil {
  1292. return false, err
  1293. }
  1294. if len(inboundIds) == 0 {
  1295. // Legacy fallback for clients that only live in the inbound JSON and
  1296. // were never normalized into client_inbounds.
  1297. _, inbound, gErr := inboundSvc.GetClientInboundByEmail(clientEmail)
  1298. if gErr != nil {
  1299. return false, gErr
  1300. }
  1301. if inbound == nil {
  1302. return false, common.NewError("Inbound Not Found For Email:", clientEmail)
  1303. }
  1304. inboundIds = []int{inbound.Id}
  1305. }
  1306. needRestart := false
  1307. found := false
  1308. // Built before any inbound is written, as in Update: only the applies
  1309. // overlap, so one node's round-trip no longer waits on the previous one.
  1310. applies := make([]inboundApply, 0, len(inboundIds))
  1311. for _, ibId := range inboundIds {
  1312. inbound, gErr := inboundSvc.GetInbound(ibId)
  1313. if gErr != nil {
  1314. return needRestart, gErr
  1315. }
  1316. var settings map[string]any
  1317. if uErr := json.Unmarshal([]byte(inbound.Settings), &settings); uErr != nil {
  1318. return needRestart, uErr
  1319. }
  1320. clients, _ := settings["clients"].([]any)
  1321. // UpdateInboundClient expects a single-client payload, so keep only the
  1322. // matching entry in the scratch copy; it splices the result back into
  1323. // the inbound's full client list itself.
  1324. var newClients []any
  1325. for client_index := range clients {
  1326. c, ok := clients[client_index].(map[string]any)
  1327. if !ok {
  1328. continue
  1329. }
  1330. if c["email"] == clientEmail {
  1331. mutate(c)
  1332. c["updated_at"] = time.Now().Unix() * 1000
  1333. newClients = append(newClients, any(c))
  1334. }
  1335. }
  1336. if len(newClients) == 0 {
  1337. continue
  1338. }
  1339. found = true
  1340. settings["clients"] = newClients
  1341. modifiedSettings, mErr := json.MarshalIndent(settings, "", " ")
  1342. if mErr != nil {
  1343. return needRestart, mErr
  1344. }
  1345. inbound.Settings = string(modifiedSettings)
  1346. data := inbound
  1347. applies = append(applies, inboundApply{id: ibId, run: func() (bool, error) {
  1348. return s.UpdateInboundClient(inboundSvc, data, clientEmail)
  1349. }})
  1350. }
  1351. if !found {
  1352. return needRestart, common.NewError("Client Not Found For Email:", clientEmail)
  1353. }
  1354. nr, applyErr := fanoutInboundApplies(applies)
  1355. return needRestart || nr, applyErr
  1356. }
  1357. func (s *ClientService) ResetClientIpLimitByEmail(inboundSvc *InboundService, clientEmail string, count int) (bool, error) {
  1358. return s.applyClientFieldByEmail(inboundSvc, clientEmail, func(c map[string]any) {
  1359. c["limitIp"] = count
  1360. })
  1361. }
  1362. func (s *ClientService) ResetClientExpiryTimeByEmail(inboundSvc *InboundService, clientEmail string, expiry_time int64) (bool, error) {
  1363. return s.applyClientFieldByEmail(inboundSvc, clientEmail, func(c map[string]any) {
  1364. c["expiryTime"] = expiry_time
  1365. })
  1366. }
  1367. func (s *ClientService) ResetClientTrafficLimitByEmail(inboundSvc *InboundService, clientEmail string, totalGB int) (bool, error) {
  1368. if totalGB < 0 {
  1369. return false, common.NewError("totalGB must be >= 0")
  1370. }
  1371. return s.applyClientFieldByEmail(inboundSvc, clientEmail, func(c map[string]any) {
  1372. c["totalGB"] = totalGB * 1024 * 1024 * 1024
  1373. })
  1374. }