mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-09-30 19:22:10 +03:00
9672249edb
* feat(clients): add calendar weekly renewal and schedule previews Expose fixed-day, calendar-weekly, calendar-monthly, and disabled renewal through one shared selector in individual and bulk client forms. Store the weekly weekday separately (Monday 1 through Sunday 7) and use panel-local calendar dates rather than a fixed 168-hour duration. Resolve skipped or repeated midnights to the first valid instant of the selected date, and skip an entirely nonexistent calendar date rather than changing the weekday. Reuse the existing renewal writer and share its boundary alignment and per-period catch-up calculation with an authenticated, read-only preview. Keep monthly precedence for legacy records, fixed-day interval semantics, maximum renewal allowances, first-use durations, and operator-disabled settings unchanged. Selecting a mode does not rewrite an existing cutoff; an unset calendar cutoff requires an explicit action to choose the first. The last-valid-second preview uses the stored exclusive expiry, even when the billing calculation aligns a legacy last-second cutoff up to midnight. Carry weekly schedules through client persistence, paging, enable toggles, inbound settings, and node traffic reconciliation. Migrate missing or nullable weekday columns to disabled by default without altering existing limits, and include the new isolated-schema PostgreSQL regression in the live CI gate. Regenerate API contracts and reference documentation, add lifecycle and form regressions, and document timezone, quota-reset, and upgrade considerations. All participating nodes must be upgraded before weekly mode is enabled; older binaries ignore the new field. Independent periodic traffic resets and the optional month-end subscription-header display are not changed. * fix(clients): validate renewal schedules across inbound write paths Reject conflicting weekly/interval/monthly schedules and out-of-range weekdays on inbound creation and edits, legacy one-client apply paths, record/link synchronization, and traffic metadata writes. Validate imported traffic snapshots as well, before any inbound or client is persisted, so an inbound API cannot create a client that the clients page cannot toggle. Merge a weekly-related schedule as one timestamp-selected tuple rather than filling its zero fields from another renewal mode. Preserve empty migration snapshots and the existing non-weekly monthly/interval merge semantics. Renewal caps, counters, credentials, and deadlines are unchanged. Add regressions for nine write paths, unchanged records and runtime calls after rejection, valid inbound clients remaining editable, and duplicate record merges between individually valid renewal modes. * docs(clients): clarify depleted-client deletion risks on downgrade Explain in English and Chinese that older versions not only stop weekly renewal: their depleted-client cleanup can delete a weekly-only client once its expiry or quota is exhausted. This is conditional on cleanup, not an automatic deletion caused by downgrade itself. Recommend backing up and converting weekly schedules to a mode supported by every participating version before rollback, and avoiding cleanup while mixed versions or unconverted clients remain. Merely disabling weekly renewal does not restore the old binary's missing purge protection. * fix(clients): bound weekly renewal date searches Limit the search for a valid weekly calendar date to eight candidates so an unusual timezone cannot monopolize the single traffic writer. Exhaustion returns the original instant, allowing the existing catch-up forward-progress guard to stop without advancing expiry, consuming an allowance, resetting traffic, or falling back to a fixed-duration schedule that can drift. Reject a non-future calendar suggestion in the read-only preview instead of offering an immediately expired initial cutoff. Also report failed weekly catch-up as a search error when allowances remain, not as cap exhaustion. Existing preview errors use the form's current warning; no API schema or locale changes are needed. Exercise exhaustion with a synthetic valid TZif containing twelve skipped Sundays. This fault-injection case was red without the bound; it is not a claim that a production IANA timezone was observed hanging. Keep the Havana and Apia regressions for real skipped/repeated midnights and absent dates. * fix(tests): isolate weekly renewal preview timezone Stop the weekly search regression from replacing process-global time.Local. CI caught that assignment and its cleanup racing with background timer reads through time.Now, even though the top-level tests do not use t.Parallel. Pass the timezone and current instant into the unchanged preview calculation. The public service still validates the request and resolves the panel timezone; API responses, renewal accounting, and persisted client data are unchanged. Use fixed dates for both suggestion and catch-up exhaustion, removing the test's dependency on today's date and its unnecessary database setup. Keep a bounded-lifetime background clock reader to expose future global-timezone mutations under the existing race gate rather than disabling that check. * ci: retrigger PR checks Create an empty commit to request a fresh pull-request CI run after release dependency downloads failed with network errors. No source, dependency, or workflow changes are included. Retry the existing checks without bypassing them. * ci: retry PR checks and record deferred download hardening Request another CI run after the amd64 release job compiled successfully but failed during dependency fetching with exit code 4 (network failure). Record possible follow-up improvements for the Linux release fetch helper: - Print each download URL and destination, and preserve error details. - Reuse the existing curl configuration with up to five retries; add connection and per-attempt timeouts and a bounded retry window. - Download to a temporary file and promote it to the final filename only after a successful, non-empty transfer. Keep the job failing if downloads ultimately fail. - Validate successful downloads, recovery after a temporary failure, and correct failure after persistent errors before shipping such a change. These improvements are intentionally deferred, not implemented or tested by this commit. This commit is empty: renewal logic, dependencies, workflow configuration, check requirements, and TLS verification remain unchanged. --------- Co-authored-by: JacktheRanger <219502738+JacktheRanger@users.noreply.github.com>
1475 lines
45 KiB
Go
1475 lines
45 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"maps"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/mhsanaei/3x-ui/v3/internal/database"
|
|
"github.com/mhsanaei/3x-ui/v3/internal/database/model"
|
|
"github.com/mhsanaei/3x-ui/v3/internal/logger"
|
|
"github.com/mhsanaei/3x-ui/v3/internal/util/common"
|
|
"github.com/mhsanaei/3x-ui/v3/internal/util/random"
|
|
"github.com/mhsanaei/3x-ui/v3/internal/web/runtime"
|
|
"github.com/mhsanaei/3x-ui/v3/internal/xray"
|
|
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
// droppedClientNeedsRestart is what restartXrayOnClientDisable asks for: the core
|
|
// API drops the credential only, so a live session needs the process replaced.
|
|
func droppedClientNeedsRestart() bool {
|
|
on, err := (&SettingService{}).GetRestartXrayOnClientDisable()
|
|
if err != nil {
|
|
logger.Warning("get RestartXrayOnClientDisable failed:", err)
|
|
}
|
|
return on
|
|
}
|
|
|
|
func sameClientConfigExceptUpdatedAt(a, b map[string]any) bool {
|
|
aa := maps.Clone(a)
|
|
bb := maps.Clone(b)
|
|
delete(aa, "updated_at")
|
|
delete(bb, "updated_at")
|
|
an, aerr := json.Marshal(aa)
|
|
bn, berr := json.Marshal(bb)
|
|
return aerr == nil && berr == nil && string(an) == string(bn)
|
|
}
|
|
|
|
// advancePushedInbound advances the node's reconcile-skip fingerprint from the
|
|
// pre-edit settings to the saved ones after every per-client push succeeded.
|
|
func advancePushedInbound(rt runtime.Runtime, prevSettings string, ib *model.Inbound) {
|
|
rem, ok := rt.(*runtime.Remote)
|
|
if !ok {
|
|
return
|
|
}
|
|
prev := *ib
|
|
prev.Settings = prevSettings
|
|
rem.AdvancePushedInbound(&prev, ib)
|
|
}
|
|
|
|
// delInboundClients removes several clients from a single inbound in one pass:
|
|
// one settings rewrite, one runtime sweep, one Save and one link delta for the
|
|
// whole batch, instead of repeating the full per-client cycle. It mirrors the
|
|
// semantics of DelInboundClientByEmail for each removed client. needRestart is
|
|
// the OR across all removals.
|
|
func (s *ClientService) delInboundClients(inboundSvc *InboundService, inboundId int, recs []*model.ClientRecord, keepTraffic bool) (bool, error) {
|
|
if len(recs) == 0 {
|
|
return false, nil
|
|
}
|
|
defer lockInbound(inboundId).Unlock()
|
|
|
|
oldInbound, err := inboundSvc.GetInbound(inboundId)
|
|
if err != nil {
|
|
logger.Error("Load Old Data Error")
|
|
return false, err
|
|
}
|
|
|
|
var settings map[string]any
|
|
if err := json.Unmarshal([]byte(oldInbound.Settings), &settings); err != nil {
|
|
return false, err
|
|
}
|
|
|
|
// Match by email — the client's stable identity (see Delete). Removes every
|
|
// entry carrying a wanted email, independent of credential drift.
|
|
wanted := make(map[string]struct{}, len(recs))
|
|
for _, rec := range recs {
|
|
if rec.Email != "" {
|
|
wanted[rec.Email] = struct{}{}
|
|
}
|
|
}
|
|
|
|
interfaceClients, ok := settings["clients"].([]any)
|
|
if !ok {
|
|
return false, common.NewError("invalid clients format in inbound settings")
|
|
}
|
|
|
|
type removedClient struct {
|
|
email string
|
|
needApiDel bool
|
|
}
|
|
removed := make([]removedClient, 0, len(wanted))
|
|
newClients := make([]any, 0, len(interfaceClients))
|
|
for _, client := range interfaceClients {
|
|
c, ok := client.(map[string]any)
|
|
if !ok {
|
|
newClients = append(newClients, client)
|
|
continue
|
|
}
|
|
email, _ := c["email"].(string)
|
|
if _, hit := wanted[email]; hit && email != "" {
|
|
enable, _ := c["enable"].(bool)
|
|
removed = append(removed, removedClient{email: email, needApiDel: enable})
|
|
continue
|
|
}
|
|
newClients = append(newClients, client)
|
|
}
|
|
|
|
if len(removed) == 0 {
|
|
return false, nil
|
|
}
|
|
|
|
db := database.GetDB()
|
|
newClients = compactOrphans(db, newClients)
|
|
if newClients == nil {
|
|
newClients = []any{}
|
|
}
|
|
settings["clients"] = newClients
|
|
newSettings, err := json.MarshalIndent(settings, "", " ")
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
prevSettings := oldInbound.Settings
|
|
oldInbound.Settings = string(newSettings)
|
|
|
|
var sharedSet map[string]bool
|
|
if !keepTraffic {
|
|
removedEmails := make([]string, 0, len(removed))
|
|
for _, r := range removed {
|
|
if r.email != "" {
|
|
removedEmails = append(removedEmails, r.email)
|
|
}
|
|
}
|
|
var sharedErr error
|
|
sharedSet, sharedErr = inboundSvc.emailsUsedByOtherInbounds(removedEmails, inboundId)
|
|
if sharedErr != nil {
|
|
return false, sharedErr
|
|
}
|
|
}
|
|
|
|
needRestart := false
|
|
|
|
// Read each client's live state before the DB write (DelClientStat would
|
|
// erase the enable flag we need to decide on a runtime removal).
|
|
type delTarget struct {
|
|
email string
|
|
emailShared bool
|
|
notDepleted bool
|
|
needApiDel bool
|
|
}
|
|
targets := make([]delTarget, 0, len(removed))
|
|
for _, r := range removed {
|
|
email := r.email
|
|
emailShared := sharedSet[strings.ToLower(strings.TrimSpace(email))]
|
|
notDepleted := false
|
|
if len(email) > 0 {
|
|
var enables []bool
|
|
if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email).Limit(1).Pluck("enable", &enables).Error; err != nil {
|
|
logger.Error("Get stats error")
|
|
return needRestart, err
|
|
}
|
|
notDepleted = len(enables) > 0 && enables[0]
|
|
}
|
|
targets = append(targets, delTarget{email: email, emailShared: emailShared, notDepleted: notDepleted, needApiDel: r.needApiDel})
|
|
}
|
|
|
|
// Persist the batch deletion atomically, serialized against the traffic poll
|
|
// to avoid the cross-transaction lock-order deadlock (runSerializedTx).
|
|
if txErr := runSerializedTx(func(tx *gorm.DB) error {
|
|
for _, t := range targets {
|
|
if t.emailShared || keepTraffic {
|
|
continue
|
|
}
|
|
if e := inboundSvc.DelClientIPs(tx, t.email); e != nil {
|
|
logger.Error("Error in delete client IPs")
|
|
return e
|
|
}
|
|
if len(t.email) > 0 {
|
|
if e := inboundSvc.DelClientStat(tx, t.email); e != nil {
|
|
logger.Error("Delete stats Data Error")
|
|
return e
|
|
}
|
|
}
|
|
}
|
|
if e := tx.Save(oldInbound).Error; e != nil {
|
|
return e
|
|
}
|
|
detached := make([]string, 0, len(targets))
|
|
for _, t := range targets {
|
|
if t.email != "" {
|
|
detached = append(detached, t.email)
|
|
}
|
|
}
|
|
if err := s.ApplyInboundClientDelta(tx, inboundId, nil, detached); err != nil {
|
|
return err
|
|
}
|
|
if oldInbound.NodeID != nil {
|
|
return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
|
|
}
|
|
return nil
|
|
}); txErr != nil {
|
|
return needRestart, txErr
|
|
}
|
|
|
|
// Resolve the node push plan once for the whole batch instead of per email.
|
|
var nodeRt runtime.Runtime
|
|
nodePush := false
|
|
if oldInbound.NodeID != nil {
|
|
rt, push, _, perr := inboundSvc.nodePushPlan(oldInbound)
|
|
if perr != nil {
|
|
return needRestart, perr
|
|
}
|
|
nodeRt, nodePush = rt, push
|
|
// Large batches collapse into one reconcile push rather than M deletes.
|
|
if nodePush && len(targets) > nodeBulkPushThreshold {
|
|
nodePush = false
|
|
}
|
|
}
|
|
|
|
// Apply runtime deletes after commit — outside the serialized writer so a
|
|
// slow node call can't stall traffic accounting.
|
|
nodePushFailed := false
|
|
for _, t := range targets {
|
|
if len(t.email) == 0 {
|
|
continue
|
|
}
|
|
if oldInbound.NodeID == nil {
|
|
if t.needApiDel && t.notDepleted {
|
|
rt, rterr := inboundSvc.runtimeFor(oldInbound)
|
|
if rterr != nil {
|
|
needRestart = true
|
|
} else if err1 := rt.RemoveUser(context.Background(), oldInbound, t.email); err1 != nil {
|
|
if !strings.Contains(err1.Error(), fmt.Sprintf("User %s not found.", t.email)) {
|
|
needRestart = true
|
|
}
|
|
}
|
|
}
|
|
} else if nodePush && !nodePushFailed {
|
|
// First failure ends the batch push; the reconcile converges the rest.
|
|
ctx, cancel := nodePushContext()
|
|
err1 := nodeRt.DeleteUser(ctx, oldInbound, t.email)
|
|
cancel()
|
|
if err1 != nil {
|
|
logger.Warning("Error in deleting client on", nodeRt.Name(), ":", err1)
|
|
nodePushFailed = true
|
|
}
|
|
}
|
|
}
|
|
if nodePush && !nodePushFailed {
|
|
advancePushedInbound(nodeRt, prevSettings, oldInbound)
|
|
}
|
|
|
|
return needRestart, nil
|
|
}
|
|
|
|
// otherTunnelAllowedIPs maps every AllowedIPs entry claimed on another
|
|
// WireGuard/AmneziaWG inbound to a description of which one holds it: the
|
|
// per-inbound defaulters only check their own client list, so two inbounds
|
|
// sharing a subnet could otherwise hand out the same address. Disabled
|
|
// siblings count too, keeping their addresses reserved for a later re-enable.
|
|
//
|
|
// selfEmails skips this identity's own entries. Email is globally unique, so a
|
|
// match there is never a real collision -- and Attach deliberately reuses one
|
|
// address across every inbound it attaches the identity to.
|
|
func (s *ClientService) otherTunnelAllowedIPs(db *gorm.DB, inboundSvc *InboundService, excludeID int, selfEmails map[string]struct{}) (map[string]string, error) {
|
|
var inbounds []*model.Inbound
|
|
err := db.Model(model.Inbound{}).
|
|
Where("protocol IN ? AND id != ?", []model.Protocol{model.WireGuard, model.AmneziaWG}, excludeID).
|
|
Find(&inbounds).Error
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
used := make(map[string]string)
|
|
for _, ib := range inbounds {
|
|
clients, cErr := inboundSvc.GetClients(ib)
|
|
if cErr != nil {
|
|
continue
|
|
}
|
|
name := ib.Remark
|
|
if name == "" {
|
|
name = ib.Tag
|
|
}
|
|
label := fmt.Sprintf("inbound '%s' (#%d)", name, ib.Id)
|
|
for _, c := range clients {
|
|
if _, self := selfEmails[strings.ToLower(c.Email)]; self {
|
|
continue
|
|
}
|
|
for _, addr := range c.AllowedIPs {
|
|
used[addr] = label
|
|
}
|
|
}
|
|
}
|
|
return used, nil
|
|
}
|
|
|
|
func (s *ClientService) checkEmailsExistForClients(inboundSvc *InboundService, clients []model.Client) (string, error) {
|
|
emailSubIDs, err := inboundSvc.emailSubIDsForClients(clients)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
seen := make(map[string]string, len(clients))
|
|
for _, client := range clients {
|
|
if client.Email == "" {
|
|
continue
|
|
}
|
|
key := strings.ToLower(client.Email)
|
|
if prev, ok := seen[key]; ok {
|
|
if prev != client.SubID || client.SubID == "" {
|
|
return client.Email, nil
|
|
}
|
|
continue
|
|
}
|
|
seen[key] = client.SubID
|
|
if existingSub, ok := emailSubIDs[key]; ok {
|
|
if client.SubID == "" || existingSub == "" || existingSub != client.SubID {
|
|
return client.Email, nil
|
|
}
|
|
}
|
|
}
|
|
return "", nil
|
|
}
|
|
|
|
func (s *ClientService) AddInboundClient(inboundSvc *InboundService, data *model.Inbound) (bool, error) {
|
|
defer lockInbound(data.Id).Unlock()
|
|
|
|
clients, err := inboundSvc.GetClients(data)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if err := validateClientsRenewal(clients); err != nil {
|
|
return false, err
|
|
}
|
|
|
|
var settings map[string]any
|
|
err = json.Unmarshal([]byte(data.Settings), &settings)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
|
|
interfaceClients := settings["clients"].([]any)
|
|
nowTs := time.Now().Unix() * 1000
|
|
for i := range interfaceClients {
|
|
if cm, ok := interfaceClients[i].(map[string]any); ok {
|
|
if _, ok2 := cm["created_at"]; !ok2 {
|
|
cm["created_at"] = nowTs
|
|
}
|
|
cm["updated_at"] = nowTs
|
|
existingSub, _ := cm["subId"].(string)
|
|
if strings.TrimSpace(existingSub) == "" {
|
|
cm["subId"] = random.NumLower(16)
|
|
}
|
|
interfaceClients[i] = cm
|
|
}
|
|
}
|
|
existEmail, err := s.checkEmailsExistForClients(inboundSvc, clients)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if existEmail != "" {
|
|
return false, common.NewError("Duplicate email:", existEmail)
|
|
}
|
|
|
|
oldInbound, err := inboundSvc.GetInbound(data.Id)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
|
|
existingClients, err := inboundSvc.GetClients(oldInbound)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
|
|
// A client already on this inbound is skipped instead of appended again:
|
|
// checkEmailsExistForClients exempts a matching subId so one identity can
|
|
// live on several inbounds, which let retried or raced adds duplicate the
|
|
// same email inside a single settings array (#5770). clients and
|
|
// interfaceClients are parsed from the same data.Settings array, so they
|
|
// stay index-aligned while filtering.
|
|
if len(existingClients) > 0 && len(clients) > 0 {
|
|
existingEmails := make(map[string]struct{}, len(existingClients))
|
|
for _, c := range existingClients {
|
|
if c.Email != "" {
|
|
existingEmails[strings.ToLower(c.Email)] = struct{}{}
|
|
}
|
|
}
|
|
keptClients := make([]model.Client, 0, len(clients))
|
|
keptWire := make([]any, 0, len(interfaceClients))
|
|
for i, c := range clients {
|
|
if c.Email != "" {
|
|
if _, dup := existingEmails[strings.ToLower(c.Email)]; dup {
|
|
continue
|
|
}
|
|
}
|
|
keptClients = append(keptClients, c)
|
|
if i < len(interfaceClients) {
|
|
keptWire = append(keptWire, interfaceClients[i])
|
|
}
|
|
}
|
|
if len(keptClients) == 0 {
|
|
return false, nil
|
|
}
|
|
clients = keptClients
|
|
interfaceClients = keptWire
|
|
}
|
|
|
|
var selfEmails map[string]struct{}
|
|
if oldInbound.Protocol == model.WireGuard || oldInbound.Protocol == model.AmneziaWG {
|
|
selfEmails = make(map[string]struct{}, len(clients))
|
|
for _, c := range clients {
|
|
if c.Email != "" {
|
|
selfEmails[strings.ToLower(c.Email)] = struct{}{}
|
|
}
|
|
}
|
|
crossUsed, cErr := s.otherTunnelAllowedIPs(database.GetDB(), inboundSvc, oldInbound.Id, selfEmails)
|
|
if cErr != nil {
|
|
return false, cErr
|
|
}
|
|
if oldInbound.Protocol == model.WireGuard {
|
|
if dErr := defaultWireguardClients(oldInbound.Settings, existingClients, clients, interfaceClients, crossUsed); dErr != nil {
|
|
return false, dErr
|
|
}
|
|
}
|
|
if oldInbound.Protocol == model.AmneziaWG {
|
|
if dErr := defaultAmneziaWGClients(oldInbound.Settings, existingClients, clients, interfaceClients, crossUsed); dErr != nil {
|
|
return false, dErr
|
|
}
|
|
}
|
|
}
|
|
|
|
var portCtx portConflictContext
|
|
if oldInbound.Protocol == model.AmneziaWG {
|
|
portCtx, err = inboundSvc.loadPortConflictContext(database.GetDB())
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
}
|
|
for _, client := range clients {
|
|
if strings.TrimSpace(client.Email) == "" {
|
|
return false, common.NewError("client email is required")
|
|
}
|
|
switch oldInbound.Protocol {
|
|
case "trojan":
|
|
if client.Password == "" {
|
|
return false, common.NewError("empty client ID")
|
|
}
|
|
case "shadowsocks":
|
|
if client.Email == "" {
|
|
return false, common.NewError("empty client ID")
|
|
}
|
|
case "hysteria":
|
|
if client.Auth == "" {
|
|
return false, common.NewError("empty client ID")
|
|
}
|
|
case "wireguard", "amneziawg":
|
|
if client.PublicKey == "" {
|
|
return false, common.NewError("wireguard client requires a key")
|
|
}
|
|
case "mtproto":
|
|
if client.Secret == "" {
|
|
return false, common.NewError("mtproto client requires a secret")
|
|
}
|
|
if client.AdTag != "" && !model.ValidMtprotoAdTag(client.AdTag) {
|
|
return false, common.NewError("mtproto client ad tag must be 32 hex characters")
|
|
}
|
|
case "tuic":
|
|
if client.ID == "" {
|
|
return false, common.NewError("empty client ID")
|
|
}
|
|
if client.Password == "" {
|
|
return false, common.NewError("tuic client requires a password")
|
|
}
|
|
if client.Email == "" {
|
|
return false, common.NewError("empty client email")
|
|
}
|
|
default:
|
|
if client.ID == "" {
|
|
return false, common.NewError("empty client ID")
|
|
}
|
|
}
|
|
if oldInbound.Protocol == model.AmneziaWG {
|
|
if hit := inboundSvc.checkForwardedPortsConflict(portCtx, client.ForwardedPorts); hit != "" {
|
|
return false, common.NewError("amneziawg: forwardedPorts collides with", hit)
|
|
}
|
|
}
|
|
}
|
|
|
|
var oldSettings map[string]any
|
|
err = json.Unmarshal([]byte(oldInbound.Settings), &oldSettings)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
|
|
if oldInbound.Protocol == model.Shadowsocks {
|
|
applyShadowsocksClientMethod(interfaceClients, oldSettings)
|
|
}
|
|
|
|
oldClients, _ := oldSettings["clients"].([]any)
|
|
oldClients = compactOrphans(database.GetDB(), oldClients)
|
|
oldClients = append(oldClients, interfaceClients...)
|
|
|
|
oldSettings["clients"] = oldClients
|
|
|
|
newSettings, err := json.MarshalIndent(oldSettings, "", " ")
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
|
|
prevSettings := oldInbound.Settings
|
|
oldInbound.Settings = string(newSettings)
|
|
|
|
// From the stamped wire entries, not from clients: created_at / updated_at /
|
|
// subId are written onto interfaceClients above, after clients was parsed.
|
|
addedClients, err := settingsEntriesToClients(interfaceClients)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
|
|
needRestart := false
|
|
|
|
rt, push, _, perr := inboundSvc.nodePushPlan(oldInbound)
|
|
if perr != nil {
|
|
return false, perr
|
|
}
|
|
|
|
// Persist client stats + inbound atomically, serialized against the traffic
|
|
// poll to avoid the cross-transaction lock-order deadlock (runSerializedTx).
|
|
if txErr := runSerializedTx(func(tx *gorm.DB) error {
|
|
// lockInbound is per-inbound, so the pre-tx cross-inbound checks race
|
|
// concurrent writers on other inbounds — re-run them in here (#6225).
|
|
if oldInbound.Protocol == model.WireGuard || oldInbound.Protocol == model.AmneziaWG {
|
|
crossUsed, cErr := s.otherTunnelAllowedIPs(tx, inboundSvc, oldInbound.Id, selfEmails)
|
|
if cErr != nil {
|
|
return cErr
|
|
}
|
|
crossAddrs := make([]string, 0, len(crossUsed))
|
|
for addr := range crossUsed {
|
|
crossAddrs = append(crossAddrs, addr)
|
|
}
|
|
for i := range clients {
|
|
if hit := wireguardAllowedIPsCollision(clients[i].AllowedIPs, crossAddrs); hit != "" {
|
|
return common.NewError("allowedIPs entry", hit, "is already used by a client on", crossUsed[hit])
|
|
}
|
|
}
|
|
}
|
|
if oldInbound.Protocol == model.AmneziaWG {
|
|
txPortCtx, pErr := inboundSvc.loadPortConflictContext(tx)
|
|
if pErr != nil {
|
|
return pErr
|
|
}
|
|
for i := range clients {
|
|
if hit := inboundSvc.checkForwardedPortsConflict(txPortCtx, clients[i].ForwardedPorts); hit != "" {
|
|
return common.NewError("amneziawg: forwardedPorts collides with", hit)
|
|
}
|
|
}
|
|
}
|
|
for i := range clients {
|
|
if len(clients[i].Email) == 0 {
|
|
continue
|
|
}
|
|
if e := inboundSvc.AddClientStat(tx, data.Id, &clients[i]); e != nil {
|
|
return e
|
|
}
|
|
}
|
|
if e := tx.Save(oldInbound).Error; e != nil {
|
|
return e
|
|
}
|
|
if err := s.ApplyInboundClientDelta(tx, oldInbound.Id, addedClients, nil); err != nil {
|
|
return err
|
|
}
|
|
if oldInbound.NodeID != nil {
|
|
return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
|
|
}
|
|
return nil
|
|
}); txErr != nil {
|
|
return false, txErr
|
|
}
|
|
|
|
// Apply to the running runtime after commit — outside the serialized writer
|
|
// so a slow node call can't stall traffic accounting.
|
|
if oldInbound.NodeID == nil {
|
|
if !push {
|
|
needRestart = true
|
|
} else if oldInbound.Protocol == model.MTProto {
|
|
inboundSvc.applyLocalMtproto(oldInbound.Id)
|
|
} else if oldInbound.Protocol == model.AmneziaWG {
|
|
inboundSvc.applyLocalAmneziaWG(oldInbound.Id)
|
|
} else if oldInbound.Protocol == model.TUIC {
|
|
inboundSvc.applyLocalTuic(oldInbound.Id)
|
|
} else {
|
|
for _, client := range clients {
|
|
if len(client.Email) == 0 {
|
|
needRestart = true
|
|
continue
|
|
}
|
|
if !client.Enable {
|
|
continue
|
|
}
|
|
cipher := ""
|
|
if oldInbound.Protocol == "shadowsocks" {
|
|
cipher, _ = oldSettings["method"].(string)
|
|
}
|
|
err1 := rt.AddUser(context.Background(), oldInbound, map[string]any{
|
|
"email": client.Email,
|
|
"id": client.ID,
|
|
"auth": client.Auth,
|
|
"security": client.Security,
|
|
"flow": client.Flow,
|
|
"password": client.Password,
|
|
"cipher": cipher,
|
|
"publicKey": client.PublicKey,
|
|
"allowedIPs": client.AllowedIPs,
|
|
"preSharedKey": client.PreSharedKey,
|
|
"keepAlive": keepAliveStr(client.KeepAliveSeconds()),
|
|
"reverse": client.Reverse,
|
|
})
|
|
if err1 == nil {
|
|
logger.Debug("Client added on", rt.Name(), ":", client.Email)
|
|
} else {
|
|
logger.Debug("Error in adding client on", rt.Name(), ":", err1)
|
|
needRestart = true
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
// Large batches would be M sequential per-client RPCs; the inbound's saved
|
|
// settings already hold the final set, so mark dirty and let one reconcile
|
|
// push converge the node instead.
|
|
if push && len(clients) > nodeBulkPushThreshold {
|
|
push = false
|
|
}
|
|
for _, client := range clients {
|
|
// /clients/add on the node historically coerced enable=true; skip live
|
|
// push for disabled clients and leave dirty so reconcile converges.
|
|
if !client.Enable {
|
|
push = false
|
|
continue
|
|
}
|
|
if push {
|
|
ctx, cancel := nodePushContext()
|
|
err1 := rt.AddClient(ctx, oldInbound, client)
|
|
cancel()
|
|
if err1 != nil {
|
|
logger.Warning("Error in adding client on", rt.Name(), ":", err1)
|
|
push = false
|
|
}
|
|
}
|
|
}
|
|
if push {
|
|
advancePushedInbound(rt, prevSettings, oldInbound)
|
|
}
|
|
}
|
|
|
|
return needRestart, nil
|
|
}
|
|
|
|
func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *model.Inbound, oldEmail string) (bool, error) {
|
|
defer lockInbound(data.Id).Unlock()
|
|
|
|
clients, err := inboundSvc.GetClients(data)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if err := validateClientsRenewal(clients); err != nil {
|
|
return false, err
|
|
}
|
|
|
|
var settings map[string]any
|
|
err = json.Unmarshal([]byte(data.Settings), &settings)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
|
|
interfaceClients := settings["clients"].([]any)
|
|
|
|
oldInbound, err := inboundSvc.GetInbound(data.Id)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
|
|
oldClients, err := inboundSvc.GetClients(oldInbound)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
|
|
newClientId := ""
|
|
switch oldInbound.Protocol {
|
|
case "trojan":
|
|
newClientId = clients[0].Password
|
|
case "shadowsocks":
|
|
newClientId = clients[0].Email
|
|
case "hysteria":
|
|
newClientId = clients[0].Auth
|
|
case "wireguard", "amneziawg":
|
|
newClientId = clients[0].Email
|
|
case "mtproto":
|
|
newClientId = clients[0].Email
|
|
default:
|
|
newClientId = clients[0].ID
|
|
}
|
|
|
|
// Locate the client to replace by email — the client's stable identity.
|
|
// Credentials (uuid/password/auth) can drift from the inbound JSON, so they
|
|
// are never used for matching.
|
|
clientIndex := -1
|
|
for index, oldClient := range oldClients {
|
|
if strings.EqualFold(oldClient.Email, oldEmail) {
|
|
oldEmail = oldClient.Email
|
|
clientIndex = index
|
|
break
|
|
}
|
|
}
|
|
|
|
if newClientId == "" || clientIndex == -1 {
|
|
return false, common.NewError("empty client ID")
|
|
}
|
|
if strings.TrimSpace(clients[0].Email) == "" {
|
|
return false, common.NewError("client email is required")
|
|
}
|
|
if oldInbound.Protocol == model.MTProto && clients[0].AdTag != "" && !model.ValidMtprotoAdTag(clients[0].AdTag) {
|
|
return false, common.NewError("mtproto client ad tag must be 32 hex characters")
|
|
}
|
|
|
|
if clients[0].Email != oldEmail {
|
|
existEmail, err := s.checkEmailsExistForClients(inboundSvc, clients)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if existEmail != "" {
|
|
return false, common.NewError("Duplicate email:", existEmail)
|
|
}
|
|
}
|
|
|
|
// WireGuard/AmneziaWG keys are never rotated by an edit: when the incoming
|
|
// payload omits them (a metadata-only change), carry the stored credentials
|
|
// forward so the settings JSON and the running peer keep the client's identity.
|
|
if (oldInbound.Protocol == model.WireGuard || oldInbound.Protocol == model.AmneziaWG) && clientIndex >= 0 && clientIndex < len(oldClients) {
|
|
old := oldClients[clientIndex]
|
|
if clients[0].PrivateKey == "" {
|
|
clients[0].PrivateKey = old.PrivateKey
|
|
}
|
|
if clients[0].PublicKey == "" {
|
|
clients[0].PublicKey = old.PublicKey
|
|
}
|
|
if len(clients[0].AllowedIPs) == 0 {
|
|
clients[0].AllowedIPs = old.AllowedIPs
|
|
} else {
|
|
normalized, nErr := normalizeWireguardAllowedIPs(clients[0].AllowedIPs)
|
|
if nErr != nil {
|
|
return false, nErr
|
|
}
|
|
if len(normalized) == 0 {
|
|
clients[0].AllowedIPs = old.AllowedIPs
|
|
} else {
|
|
peers := make([]string, 0, len(oldClients))
|
|
for i := range oldClients {
|
|
if i == clientIndex {
|
|
continue
|
|
}
|
|
peers = append(peers, oldClients[i].AllowedIPs...)
|
|
}
|
|
if hit := wireguardAllowedIPsCollision(normalized, peers); hit != "" {
|
|
return false, common.NewError("wireguard: allowedIPs entry already used by another client:", hit)
|
|
}
|
|
clients[0].AllowedIPs = normalized
|
|
}
|
|
}
|
|
if clients[0].PreSharedKey == "" {
|
|
clients[0].PreSharedKey = old.PreSharedKey
|
|
}
|
|
if clients[0].KeepAlive == nil {
|
|
clients[0].KeepAlive = old.KeepAlive
|
|
}
|
|
// ForwardedPorts is AmneziaWG-only (WireGuard's own inbound never
|
|
// reads it), same carry-forward reasoning as the fields above: a
|
|
// partial edit (e.g. a Telegram-bot enable/expiry toggle, or an API
|
|
// call that omits the field) must not silently drop a client's
|
|
// existing port-forwarding spec.
|
|
if oldInbound.Protocol == model.AmneziaWG && clients[0].ForwardedPorts == "" {
|
|
clients[0].ForwardedPorts = old.ForwardedPorts
|
|
}
|
|
}
|
|
if oldInbound.Protocol == model.AmneziaWG {
|
|
portCtx, err := inboundSvc.loadPortConflictContext(database.GetDB())
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if hit := inboundSvc.checkForwardedPortsConflict(portCtx, clients[0].ForwardedPorts); hit != "" {
|
|
return false, common.NewError("amneziawg: forwardedPorts collides with", hit)
|
|
}
|
|
}
|
|
|
|
var oldSettings map[string]any
|
|
err = json.Unmarshal([]byte(oldInbound.Settings), &oldSettings)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
settingsClients, _ := oldSettings["clients"].([]any)
|
|
var preservedCreated any
|
|
var preservedSubID string
|
|
var oldClientMap map[string]any
|
|
if clientIndex >= 0 && clientIndex < len(settingsClients) {
|
|
if oldMap, ok := settingsClients[clientIndex].(map[string]any); ok {
|
|
oldClientMap = oldMap
|
|
if v, ok2 := oldMap["created_at"]; ok2 {
|
|
preservedCreated = v
|
|
}
|
|
preservedSubID, _ = oldMap["subId"].(string)
|
|
}
|
|
}
|
|
if oldInbound.Protocol == model.Shadowsocks {
|
|
applyShadowsocksClientMethod(interfaceClients, oldSettings)
|
|
}
|
|
if len(interfaceClients) > 0 {
|
|
if newMap, ok := interfaceClients[0].(map[string]any); ok {
|
|
if preservedCreated == nil {
|
|
preservedCreated = time.Now().Unix() * 1000
|
|
}
|
|
newMap["created_at"] = preservedCreated
|
|
newSub, _ := newMap["subId"].(string)
|
|
if strings.TrimSpace(newSub) == "" {
|
|
if strings.TrimSpace(preservedSubID) != "" {
|
|
newMap["subId"] = preservedSubID
|
|
} else {
|
|
newMap["subId"] = random.NumLower(16)
|
|
}
|
|
}
|
|
if v, ok2 := newMap["subId"].(string); ok2 {
|
|
clients[0].SubID = v
|
|
}
|
|
if oldInbound.Protocol == model.WireGuard || oldInbound.Protocol == model.AmneziaWG {
|
|
newMap["privateKey"] = clients[0].PrivateKey
|
|
newMap["publicKey"] = clients[0].PublicKey
|
|
newMap["allowedIPs"] = clients[0].AllowedIPs
|
|
if clients[0].PreSharedKey != "" {
|
|
newMap["preSharedKey"] = clients[0].PreSharedKey
|
|
}
|
|
if ka := clients[0].KeepAliveSeconds(); ka > 0 {
|
|
newMap["keepAlive"] = ka
|
|
}
|
|
if oldInbound.Protocol == model.AmneziaWG && clients[0].ForwardedPorts != "" {
|
|
newMap["forwardedPorts"] = clients[0].ForwardedPorts
|
|
}
|
|
}
|
|
if oldClientMap != nil && sameClientConfigExceptUpdatedAt(oldClientMap, newMap) {
|
|
if v, ok2 := oldClientMap["updated_at"]; ok2 {
|
|
newMap["updated_at"] = v
|
|
} else {
|
|
delete(newMap, "updated_at")
|
|
}
|
|
} else {
|
|
newMap["updated_at"] = time.Now().Unix() * 1000
|
|
}
|
|
interfaceClients[0] = newMap
|
|
}
|
|
}
|
|
settingsClients[clientIndex] = interfaceClients[0]
|
|
oldSettings["clients"] = settingsClients
|
|
|
|
if oldInbound.Protocol == model.VLESS {
|
|
hasVisionFlow := false
|
|
for _, c := range settingsClients {
|
|
cm, ok := c.(map[string]any)
|
|
if !ok {
|
|
continue
|
|
}
|
|
if flow, _ := cm["flow"].(string); flow == "xtls-rprx-vision" {
|
|
hasVisionFlow = true
|
|
break
|
|
}
|
|
}
|
|
if !hasVisionFlow {
|
|
delete(oldSettings, "testseed")
|
|
}
|
|
}
|
|
|
|
newSettings, err := json.MarshalIndent(oldSettings, "", " ")
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
|
|
if string(newSettings) == oldInbound.Settings {
|
|
return false, nil
|
|
}
|
|
|
|
prevSettings := oldInbound.Settings
|
|
oldInbound.Settings = string(newSettings)
|
|
|
|
// From the stamped wire entry, not from clients[0]: created_at, the
|
|
// preserved subId and the WireGuard carry-forward land on interfaceClients.
|
|
changedClients, err := settingsEntriesToClients(interfaceClients[:1])
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
var detachEmails []string
|
|
if len(oldEmail) > 0 && oldEmail != clients[0].Email {
|
|
detachEmails = []string{oldEmail}
|
|
}
|
|
|
|
needRestart := false
|
|
|
|
// Resolve the push plan before the DB write so a node-state lookup failure
|
|
// still aborts the whole update without committing anything (it used to roll
|
|
// the transaction back). nodePushPlan only reads, so order doesn't matter.
|
|
var rt runtime.Runtime
|
|
var push bool
|
|
if len(oldEmail) > 0 {
|
|
var perr error
|
|
rt, push, _, perr = inboundSvc.nodePushPlan(oldInbound)
|
|
if perr != nil {
|
|
return false, perr
|
|
}
|
|
}
|
|
|
|
// Persist client stats + inbound atomically, serialized against the traffic
|
|
// poll to avoid the cross-transaction lock-order deadlock (runSerializedTx).
|
|
if txErr := runSerializedTx(func(tx *gorm.DB) error {
|
|
// Same re-check-inside-the-writer rule as AddInboundClient (#6225):
|
|
// the pre-tx pass can race a concurrent writer on another inbound.
|
|
if oldInbound.Protocol == model.AmneziaWG {
|
|
txPortCtx, pErr := inboundSvc.loadPortConflictContext(tx)
|
|
if pErr != nil {
|
|
return pErr
|
|
}
|
|
if hit := inboundSvc.checkForwardedPortsConflict(txPortCtx, clients[0].ForwardedPorts); hit != "" {
|
|
return common.NewError("amneziawg: forwardedPorts collides with", hit)
|
|
}
|
|
}
|
|
if len(clients[0].Email) > 0 {
|
|
if len(oldEmail) > 0 {
|
|
emailUnchanged := strings.EqualFold(oldEmail, clients[0].Email)
|
|
targetExists := int64(0)
|
|
if !emailUnchanged {
|
|
if e := tx.Model(xray.ClientTraffic{}).Where("email = ?", clients[0].Email).Count(&targetExists).Error; e != nil {
|
|
return e
|
|
}
|
|
}
|
|
if emailUnchanged || targetExists == 0 {
|
|
if e := inboundSvc.UpdateClientStat(tx, oldEmail, &clients[0]); e != nil {
|
|
return e
|
|
}
|
|
if e := inboundSvc.UpdateClientIPs(tx, oldEmail, clients[0].Email); e != nil {
|
|
return e
|
|
}
|
|
} else {
|
|
stillUsed, sErr := inboundSvc.emailUsedByOtherInbounds(oldEmail, data.Id)
|
|
if sErr != nil {
|
|
return sErr
|
|
}
|
|
if !stillUsed {
|
|
if e := inboundSvc.DelClientStat(tx, oldEmail); e != nil {
|
|
return e
|
|
}
|
|
if e := inboundSvc.DelClientIPs(tx, oldEmail); e != nil {
|
|
return e
|
|
}
|
|
}
|
|
if e := inboundSvc.UpdateClientStat(tx, clients[0].Email, &clients[0]); e != nil {
|
|
return e
|
|
}
|
|
}
|
|
} else {
|
|
if e := inboundSvc.AddClientStat(tx, data.Id, &clients[0]); e != nil {
|
|
return e
|
|
}
|
|
}
|
|
} else {
|
|
stillUsed, sErr := inboundSvc.emailUsedByOtherInbounds(oldEmail, data.Id)
|
|
if sErr != nil {
|
|
return sErr
|
|
}
|
|
if !stillUsed {
|
|
if e := inboundSvc.DelClientStat(tx, oldEmail); e != nil {
|
|
return e
|
|
}
|
|
if e := inboundSvc.DelClientIPs(tx, oldEmail); e != nil {
|
|
return e
|
|
}
|
|
}
|
|
}
|
|
|
|
if e := tx.Save(oldInbound).Error; e != nil {
|
|
return e
|
|
}
|
|
// Rename the client record in the same transaction as the settings JSON
|
|
// so no concurrent SyncInbound can see one renamed without the other.
|
|
// Byte-level compare (not EqualFold): case-only edits must rename too,
|
|
// otherwise SyncInbound's case-sensitive lookup creates a duplicate row.
|
|
if len(oldEmail) > 0 && oldEmail != clients[0].Email {
|
|
var renameTaken int64
|
|
if e := tx.Model(&model.ClientRecord{}).Where("email = ?", clients[0].Email).Count(&renameTaken).Error; e != nil {
|
|
return e
|
|
}
|
|
if renameTaken == 0 {
|
|
if e := tx.Model(&model.ClientRecord{}).Where("email = ?", oldEmail).Update("email", clients[0].Email).Error; e != nil {
|
|
return e
|
|
}
|
|
}
|
|
}
|
|
// detachEmails covers the rename the guard above refused: the old record
|
|
// keeps this inbound's link otherwise, which the full sync used to drop.
|
|
if err := s.ApplyInboundClientDelta(tx, oldInbound.Id, changedClients, detachEmails); err != nil {
|
|
return err
|
|
}
|
|
if oldInbound.NodeID != nil {
|
|
return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
|
|
}
|
|
return nil
|
|
}); txErr != nil {
|
|
return false, txErr
|
|
}
|
|
|
|
// Apply to the running runtime after the DB is committed — outside the
|
|
// serialized writer so a slow node call can't stall traffic accounting.
|
|
if len(oldEmail) > 0 {
|
|
if oldInbound.NodeID == nil {
|
|
if !push {
|
|
needRestart = true
|
|
} else if oldInbound.Protocol == model.MTProto {
|
|
inboundSvc.applyLocalMtproto(oldInbound.Id)
|
|
} else if oldInbound.Protocol == model.AmneziaWG {
|
|
inboundSvc.applyLocalAmneziaWG(oldInbound.Id)
|
|
} else if oldInbound.Protocol == model.TUIC {
|
|
inboundSvc.applyLocalTuic(oldInbound.Id)
|
|
} else {
|
|
if oldClients[clientIndex].Enable {
|
|
err1 := rt.RemoveUser(context.Background(), oldInbound, oldEmail)
|
|
if err1 == nil {
|
|
logger.Debug("Old client deleted on", rt.Name(), ":", oldEmail)
|
|
// The API removal is enough only while the client is re-added; a
|
|
// dropped one ends its session only through a restart.
|
|
if !clients[0].Enable && droppedClientNeedsRestart() {
|
|
needRestart = true
|
|
}
|
|
} else if strings.Contains(err1.Error(), fmt.Sprintf("User %s not found.", oldEmail)) {
|
|
logger.Debug("User is already deleted. Nothing to do more...")
|
|
} else {
|
|
logger.Debug("Error in deleting client on", rt.Name(), ":", err1)
|
|
needRestart = true
|
|
}
|
|
}
|
|
if clients[0].Enable {
|
|
cipher := ""
|
|
if oldInbound.Protocol == "shadowsocks" {
|
|
cipher, _ = oldSettings["method"].(string)
|
|
}
|
|
err1 := rt.AddUser(context.Background(), oldInbound, map[string]any{
|
|
"email": clients[0].Email,
|
|
"id": clients[0].ID,
|
|
"security": clients[0].Security,
|
|
"flow": clients[0].Flow,
|
|
"auth": clients[0].Auth,
|
|
"password": clients[0].Password,
|
|
"cipher": cipher,
|
|
"publicKey": clients[0].PublicKey,
|
|
"allowedIPs": clients[0].AllowedIPs,
|
|
"preSharedKey": clients[0].PreSharedKey,
|
|
"keepAlive": keepAliveStr(clients[0].KeepAliveSeconds()),
|
|
"reverse": clients[0].Reverse,
|
|
})
|
|
if err1 == nil {
|
|
logger.Debug("Client edited on", rt.Name(), ":", clients[0].Email)
|
|
} else {
|
|
logger.Debug("Error in adding client on", rt.Name(), ":", err1)
|
|
needRestart = true
|
|
}
|
|
}
|
|
}
|
|
} else if push {
|
|
ctx, cancel := nodePushContext()
|
|
err1 := rt.UpdateUser(ctx, oldInbound, oldEmail, clients[0])
|
|
cancel()
|
|
if err1 != nil {
|
|
logger.Warning("Error in updating client on", rt.Name(), ":", err1)
|
|
} else {
|
|
advancePushedInbound(rt, prevSettings, oldInbound)
|
|
}
|
|
}
|
|
} else {
|
|
logger.Debug("Client old email not found")
|
|
needRestart = true
|
|
}
|
|
|
|
return needRestart, nil
|
|
}
|
|
|
|
func (s *ClientService) DelInboundClientByEmail(inboundSvc *InboundService, inboundId int, email string, keepTraffic bool, fullDelete bool) (bool, error) {
|
|
defer lockInbound(inboundId).Unlock()
|
|
|
|
oldInbound, err := inboundSvc.GetInbound(inboundId)
|
|
if err != nil {
|
|
logger.Error("Load Old Data Error")
|
|
return false, err
|
|
}
|
|
|
|
var settings map[string]any
|
|
if err := json.Unmarshal([]byte(oldInbound.Settings), &settings); err != nil {
|
|
return false, err
|
|
}
|
|
|
|
interfaceClients, ok := settings["clients"].([]any)
|
|
if !ok {
|
|
return false, common.NewError("invalid clients format in inbound settings")
|
|
}
|
|
|
|
var newClients []any
|
|
needApiDel := false
|
|
found := false
|
|
|
|
for _, client := range interfaceClients {
|
|
c, ok := client.(map[string]any)
|
|
if !ok {
|
|
continue
|
|
}
|
|
if cEmail, ok := c["email"].(string); ok && cEmail == email {
|
|
found = true
|
|
needApiDel, _ = c["enable"].(bool)
|
|
} else {
|
|
newClients = append(newClients, client)
|
|
}
|
|
}
|
|
|
|
if !found {
|
|
return false, fmt.Errorf("%w for email: %s", ErrClientNotInInbound, email)
|
|
}
|
|
db := database.GetDB()
|
|
newClients = compactOrphans(db, newClients)
|
|
if newClients == nil {
|
|
newClients = []any{}
|
|
}
|
|
settings["clients"] = newClients
|
|
newSettings, err := json.MarshalIndent(settings, "", " ")
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
|
|
prevSettings := oldInbound.Settings
|
|
oldInbound.Settings = string(newSettings)
|
|
|
|
emailShared, err := inboundSvc.emailUsedByOtherInbounds(email, inboundId)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
|
|
needRestart := false
|
|
|
|
// Decide what to delete and the push plan before the serialized DB write —
|
|
// these are reads, and nodePushPlan failing should abort before committing.
|
|
delStat := false
|
|
if len(email) > 0 && !emailShared && !keepTraffic {
|
|
traffic, tErr := inboundSvc.GetClientTrafficByEmail(email)
|
|
if tErr != nil {
|
|
return false, tErr
|
|
}
|
|
delStat = traffic != nil
|
|
}
|
|
|
|
// The runtime user is scoped to this inbound's tag + email, so the push plan
|
|
// is resolved independently of emailShared — a sibling inbound still carrying
|
|
// the email must not suppress removing the user from this inbound's Xray.
|
|
var rt runtime.Runtime
|
|
var push bool
|
|
if len(email) > 0 && (oldInbound.NodeID != nil || needApiDel) {
|
|
r, p, _, perr := inboundSvc.nodePushPlan(oldInbound)
|
|
if perr != nil {
|
|
return false, perr
|
|
}
|
|
rt, push = r, p
|
|
}
|
|
|
|
// Persist the deletion atomically, serialized against the traffic poll to
|
|
// avoid the cross-transaction lock-order deadlock (runSerializedTx).
|
|
if txErr := runSerializedTx(func(tx *gorm.DB) error {
|
|
if !emailShared && !keepTraffic {
|
|
if e := inboundSvc.DelClientIPs(tx, email); e != nil {
|
|
logger.Error("Error in delete client IPs")
|
|
return e
|
|
}
|
|
}
|
|
if delStat {
|
|
if e := inboundSvc.DelClientStat(tx, email); e != nil {
|
|
logger.Error("Delete stats Data Error")
|
|
return e
|
|
}
|
|
}
|
|
if e := tx.Save(oldInbound).Error; e != nil {
|
|
return e
|
|
}
|
|
if err := s.ApplyInboundClientDelta(tx, inboundId, nil, []string{email}); err != nil {
|
|
return err
|
|
}
|
|
if oldInbound.NodeID != nil {
|
|
return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
|
|
}
|
|
return nil
|
|
}); txErr != nil {
|
|
return false, txErr
|
|
}
|
|
|
|
// Apply the runtime delete after commit — outside the serialized writer so a
|
|
// slow node call can't stall traffic accounting. Independent of emailShared:
|
|
// Xray users are keyed by inbound tag, so the user must be removed from this
|
|
// inbound's runtime even when the same email survives in another inbound.
|
|
if len(email) > 0 {
|
|
if oldInbound.NodeID == nil {
|
|
if oldInbound.Protocol == model.MTProto {
|
|
// mtg serves the full secret set, so any client delete re-applies
|
|
// it (removing the last client stops the sidecar) regardless of the
|
|
// client's enable state.
|
|
inboundSvc.applyLocalMtproto(oldInbound.Id)
|
|
} else if oldInbound.Protocol == model.AmneziaWG {
|
|
// Same reasoning as MTProto above: the interface config is
|
|
// regenerated from the full peer set, so any delete re-applies it.
|
|
inboundSvc.applyLocalAmneziaWG(oldInbound.Id)
|
|
} else if oldInbound.Protocol == model.TUIC {
|
|
inboundSvc.applyLocalTuic(oldInbound.Id)
|
|
} else if needApiDel {
|
|
// Local inbound: a disabled client isn't in the running Xray, so only
|
|
// a live one (needApiDel) needs an API removal.
|
|
if !push {
|
|
needRestart = true
|
|
} else if err1 := rt.RemoveUser(context.Background(), oldInbound, email); err1 == nil {
|
|
logger.Debug("Client deleted on", rt.Name(), ":", email)
|
|
needRestart = droppedClientNeedsRestart()
|
|
} else if strings.Contains(err1.Error(), fmt.Sprintf("User %s not found.", email)) {
|
|
logger.Debug("User is already deleted. Nothing to do more...")
|
|
} else {
|
|
logger.Debug("Error in deleting client on", rt.Name(), ":", email)
|
|
needRestart = true
|
|
}
|
|
}
|
|
} else {
|
|
// Node inbound: propagate the delete regardless of the enable flag —
|
|
// the node's own DB still carries a disabled client and would
|
|
// resurrect it on the next snapshot otherwise. A full client delete
|
|
// must remove the node's client record too, not just detach it from
|
|
// this inbound (#5797).
|
|
if push {
|
|
ctx, cancel := nodePushContext()
|
|
var err1 error
|
|
if fullDelete {
|
|
err1 = rt.DeleteClient(ctx, email)
|
|
} else {
|
|
err1 = rt.DeleteUser(ctx, oldInbound, email)
|
|
}
|
|
cancel()
|
|
if err1 != nil {
|
|
logger.Warning("Error in deleting client on", rt.Name(), ":", err1)
|
|
} else {
|
|
advancePushedInbound(rt, prevSettings, oldInbound)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
return needRestart, nil
|
|
}
|
|
|
|
func (s *ClientService) SetClientTelegramUserID(inboundSvc *InboundService, trafficId int, tgId int64) (bool, error) {
|
|
traffic, inbound, err := inboundSvc.GetClientInboundByTrafficID(trafficId)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if inbound == nil {
|
|
return false, common.NewError("Inbound Not Found For Traffic ID:", trafficId)
|
|
}
|
|
|
|
clientEmail := traffic.Email
|
|
|
|
oldClients, err := inboundSvc.GetClients(inbound)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
|
|
found := false
|
|
for _, oldClient := range oldClients {
|
|
if oldClient.Email == clientEmail {
|
|
found = true
|
|
break
|
|
}
|
|
}
|
|
|
|
if !found {
|
|
return false, common.NewError("Client Not Found For Email:", clientEmail)
|
|
}
|
|
|
|
var settings map[string]any
|
|
err = json.Unmarshal([]byte(inbound.Settings), &settings)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
clients := settings["clients"].([]any)
|
|
var newClients []any
|
|
for client_index := range clients {
|
|
c := clients[client_index].(map[string]any)
|
|
if c["email"] == clientEmail {
|
|
c["tgId"] = tgId
|
|
c["updated_at"] = time.Now().Unix() * 1000
|
|
newClients = append(newClients, any(c))
|
|
}
|
|
}
|
|
settings["clients"] = newClients
|
|
modifiedSettings, err := json.MarshalIndent(settings, "", " ")
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
inbound.Settings = string(modifiedSettings)
|
|
needRestart, err := s.UpdateInboundClient(inboundSvc, inbound, clientEmail)
|
|
return needRestart, err
|
|
}
|
|
|
|
func (s *ClientService) CheckIsEnabledByEmail(inboundSvc *InboundService, clientEmail string) (bool, error) {
|
|
_, inbound, err := inboundSvc.GetClientInboundByEmail(clientEmail)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if inbound == nil {
|
|
return false, common.NewError("Inbound Not Found For Email:", clientEmail)
|
|
}
|
|
|
|
clients, err := inboundSvc.GetClients(inbound)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
|
|
isEnable := false
|
|
|
|
for _, client := range clients {
|
|
if client.Email == clientEmail {
|
|
isEnable = client.Enable
|
|
break
|
|
}
|
|
}
|
|
|
|
return isEnable, err
|
|
}
|
|
|
|
func (s *ClientService) ToggleClientEnableByEmail(inboundSvc *InboundService, clientEmail string) (bool, bool, error) {
|
|
current, err := s.CheckIsEnabledByEmail(inboundSvc, clientEmail)
|
|
if err != nil {
|
|
return false, false, err
|
|
}
|
|
target := !current
|
|
needRestart, err := s.applyClientFieldByEmail(inboundSvc, clientEmail, func(c map[string]any) {
|
|
c["enable"] = target
|
|
})
|
|
if err != nil {
|
|
return false, needRestart, err
|
|
}
|
|
return target, needRestart, nil
|
|
}
|
|
|
|
func (s *ClientService) SetClientEnableByEmail(inboundSvc *InboundService, clientEmail string, enable bool) (bool, bool, error) {
|
|
current, err := s.CheckIsEnabledByEmail(inboundSvc, clientEmail)
|
|
if err != nil {
|
|
return false, false, err
|
|
}
|
|
if current == enable {
|
|
return false, false, nil
|
|
}
|
|
needRestart, err := s.applyClientFieldByEmail(inboundSvc, clientEmail, func(c map[string]any) {
|
|
c["enable"] = enable
|
|
})
|
|
if err != nil {
|
|
return false, needRestart, err
|
|
}
|
|
return true, needRestart, nil
|
|
}
|
|
|
|
// applyClientFieldByEmail loads the inbound currently hosting clientEmail,
|
|
// confirms the client exists, applies mutate to the matching client (plus a
|
|
// refreshed updated_at), and hands a single-client update payload to
|
|
// UpdateInboundClient. The rebuilt clients array intentionally contains only
|
|
// the matched client — that is the input contract UpdateInboundClient expects
|
|
// (clients[0] is the new data; clientEmail locates the row to replace). It
|
|
// backs the single-field by-email setters below.
|
|
// applyClientFieldByEmail mutates a client field on every inbound the email is
|
|
// attached to. A multi-inbound client is one logical identity: patching only
|
|
// the first inbound's JSON would leave the siblings stale, and the next
|
|
// SyncInbound over a stale sibling would revert the edit in the normalized
|
|
// records (#5039).
|
|
func (s *ClientService) applyClientFieldByEmail(inboundSvc *InboundService, clientEmail string, mutate func(c map[string]any)) (bool, error) {
|
|
inboundIds, err := s.GetInboundIdsForEmail(database.GetDB(), clientEmail)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if len(inboundIds) == 0 {
|
|
// Legacy fallback for clients that only live in the inbound JSON and
|
|
// were never normalized into client_inbounds.
|
|
_, inbound, gErr := inboundSvc.GetClientInboundByEmail(clientEmail)
|
|
if gErr != nil {
|
|
return false, gErr
|
|
}
|
|
if inbound == nil {
|
|
return false, common.NewError("Inbound Not Found For Email:", clientEmail)
|
|
}
|
|
inboundIds = []int{inbound.Id}
|
|
}
|
|
|
|
needRestart := false
|
|
found := false
|
|
// Built before any inbound is written, as in Update: only the applies
|
|
// overlap, so one node's round-trip no longer waits on the previous one.
|
|
applies := make([]inboundApply, 0, len(inboundIds))
|
|
for _, ibId := range inboundIds {
|
|
inbound, gErr := inboundSvc.GetInbound(ibId)
|
|
if gErr != nil {
|
|
return needRestart, gErr
|
|
}
|
|
|
|
var settings map[string]any
|
|
if uErr := json.Unmarshal([]byte(inbound.Settings), &settings); uErr != nil {
|
|
return needRestart, uErr
|
|
}
|
|
clients, _ := settings["clients"].([]any)
|
|
// UpdateInboundClient expects a single-client payload, so keep only the
|
|
// matching entry in the scratch copy; it splices the result back into
|
|
// the inbound's full client list itself.
|
|
var newClients []any
|
|
for client_index := range clients {
|
|
c, ok := clients[client_index].(map[string]any)
|
|
if !ok {
|
|
continue
|
|
}
|
|
if c["email"] == clientEmail {
|
|
mutate(c)
|
|
c["updated_at"] = time.Now().Unix() * 1000
|
|
newClients = append(newClients, any(c))
|
|
}
|
|
}
|
|
if len(newClients) == 0 {
|
|
continue
|
|
}
|
|
found = true
|
|
settings["clients"] = newClients
|
|
modifiedSettings, mErr := json.MarshalIndent(settings, "", " ")
|
|
if mErr != nil {
|
|
return needRestart, mErr
|
|
}
|
|
inbound.Settings = string(modifiedSettings)
|
|
data := inbound
|
|
applies = append(applies, inboundApply{id: ibId, run: func() (bool, error) {
|
|
return s.UpdateInboundClient(inboundSvc, data, clientEmail)
|
|
}})
|
|
}
|
|
|
|
if !found {
|
|
return needRestart, common.NewError("Client Not Found For Email:", clientEmail)
|
|
}
|
|
nr, applyErr := fanoutInboundApplies(applies)
|
|
return needRestart || nr, applyErr
|
|
}
|
|
|
|
func (s *ClientService) ResetClientIpLimitByEmail(inboundSvc *InboundService, clientEmail string, count int) (bool, error) {
|
|
return s.applyClientFieldByEmail(inboundSvc, clientEmail, func(c map[string]any) {
|
|
c["limitIp"] = count
|
|
})
|
|
}
|
|
|
|
func (s *ClientService) ResetClientExpiryTimeByEmail(inboundSvc *InboundService, clientEmail string, expiry_time int64) (bool, error) {
|
|
return s.applyClientFieldByEmail(inboundSvc, clientEmail, func(c map[string]any) {
|
|
c["expiryTime"] = expiry_time
|
|
})
|
|
}
|
|
|
|
func (s *ClientService) ResetClientTrafficLimitByEmail(inboundSvc *InboundService, clientEmail string, totalGB int) (bool, error) {
|
|
if totalGB < 0 {
|
|
return false, common.NewError("totalGB must be >= 0")
|
|
}
|
|
return s.applyClientFieldByEmail(inboundSvc, clientEmail, func(c map[string]any) {
|
|
c["totalGB"] = totalGB * 1024 * 1024 * 1024
|
|
})
|
|
}
|