diff --git a/internal/web/job/check_client_ip_job.go b/internal/web/job/check_client_ip_job.go index 4773515f5..f9c39315a 100644 --- a/internal/web/job/check_client_ip_job.go +++ b/internal/web/job/check_client_ip_job.go @@ -820,11 +820,43 @@ func (j *CheckClientIpJob) disconnectClientTemporarily(inbound *model.Inbound, c // only widens the window in which new handshakes fail. time.Sleep(100 * time.Millisecond) + // clientConfig is the scan's snapshot: a disable or delete committed since then + // found no user to remove (or restarted the core) and will not come back for it. + if !j.clientStillServed(inbound.Id, clientEmail) { + logger.Infof("[LIMIT_IP] Not restoring %s: disabled or removed during its temporary disconnect", clientEmail) + return + } + // Re-add user to allow new connections err = xrayAPI.AddUser(protocol, inbound.Tag, clientConfig) if err != nil { logger.Warningf("[LIMIT_IP] Failed to re-add user %s: %v", clientEmail, err) + return } + + // Asked again: a commit can land between the check and the re-add. Writers touch + // the core only after they commit, so one this check misses removes the user itself. + if !j.clientStillServed(inbound.Id, clientEmail) { + logger.Infof("[LIMIT_IP] Removing %s again: disabled or removed during its temporary disconnect", clientEmail) + if err := xrayAPI.RemoveUser(inbound.Tag, clientEmail); err != nil { + logger.Warningf("[LIMIT_IP] Failed to remove user %s again: %v", clientEmail, err) + } + } +} + +// clientStillServed is GetXrayConfig's rule for handing a client to the core: linked to +// the enabled inbound, enabled, not depleted per client_traffics. A failed read is "no". +func (j *CheckClientIpJob) clientStillServed(inboundId int, clientEmail string) bool { + var served int64 + err := database.GetDB().Table("client_inbounds"). + Joins("JOIN clients ON clients.id = client_inbounds.client_id"). + Joins("JOIN inbounds ON inbounds.id = client_inbounds.inbound_id"). + Where("client_inbounds.inbound_id = ? AND inbounds.enable = ?", inboundId, true). + Where("clients.email = ? AND clients.enable = ?", clientEmail, true). + Where("NOT EXISTS (SELECT 1 FROM client_traffics WHERE client_traffics.email = clients.email AND client_traffics.enable = ?)", false). + Count(&served).Error + j.checkError(err) + return err == nil && served > 0 } // resolveXrayAPIPort returns the API inbound port from running config, then template config, then default. diff --git a/internal/web/job/limit_ip_disconnect_test.go b/internal/web/job/limit_ip_disconnect_test.go index f305fd652..e08243edc 100644 --- a/internal/web/job/limit_ip_disconnect_test.go +++ b/internal/web/job/limit_ip_disconnect_test.go @@ -1,11 +1,24 @@ package job import ( + "context" + "fmt" + "net" + "os" + "path/filepath" + "slices" "strings" + "sync" "testing" + "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/xray" + + "github.com/xtls/xray-core/app/proxyman/command" + "google.golang.org/grpc" + "gorm.io/gorm" ) // A reverse client has to be left alone: the cycle drops its reverse outbound @@ -73,3 +86,104 @@ func TestDisconnectClientTemporarilyAllowsHysteria(t *testing.T) { t.Fatal("expected a remove attempt against the Xray API for hysteria") } } + +// fakeCoreAPI stands in for the core's HandlerService: it records the user +// operations it is asked for and runs onOp inside each, where another job would act. +type fakeCoreAPI struct { + command.UnimplementedHandlerServiceServer + mu sync.Mutex + ops []string + onOp func(op string) +} + +func (f *fakeCoreAPI) AlterInbound(_ context.Context, req *command.AlterInboundRequest) (*command.AlterInboundResponse, error) { + operation, err := req.Operation.GetInstance() + if err != nil { + return nil, err + } + op := "add" + if _, ok := operation.(*command.RemoveUserOperation); ok { + op = "remove" + } + f.mu.Lock() + defer f.mu.Unlock() + f.ops = append(f.ops, op) + f.onOp(op) + return &command.AlterInboundResponse{}, nil +} + +func startFakeCoreAPI(t *testing.T, core *fakeCoreAPI) { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + server := grpc.NewServer() + command.RegisterHandlerServiceServer(server, core) + go func() { _ = server.Serve(listener) }() + t.Cleanup(server.Stop) + + binDir := t.TempDir() + t.Setenv("XUI_BIN_FOLDER", binDir) + config := fmt.Sprintf(`{"inbounds":[{"tag":"api","port":%d}]}`, listener.Addr().(*net.TCPAddr).Port) + if err := os.WriteFile(filepath.Join(binDir, "config.json"), []byte(config), 0o600); err != nil { + t.Fatal(err) + } +} + +// The re-add hands the core the scan's snapshot. Whatever took the client out of +// service since then has already been to the core, so the account must not stay. +func TestDisconnectClientTemporarilyDoesNotRestoreDroppedClient(t *testing.T) { + const email = "readd-probe" + trafficJobDisables := func(db *gorm.DB, _ *model.Inbound) error { + return db.Model(&xray.ClientTraffic{}).Where("email = ?", email).Update("enable", false).Error + } + tests := []struct { + name string + during string + drop func(db *gorm.DB, inbound *model.Inbound) error + wantOps []string + }{ + {name: "still served", wantOps: []string{"remove", "add"}}, + {name: "expired while removed", during: "remove", drop: trafficJobDisables, wantOps: []string{"remove"}}, + {name: "expired while being re-added", during: "add", drop: trafficJobDisables, wantOps: []string{"remove", "add", "remove"}}, + {name: "disabled by the operator", during: "remove", wantOps: []string{"remove"}, drop: func(db *gorm.DB, _ *model.Inbound) error { + return db.Model(&model.ClientRecord{}).Where("email = ?", email).Update("enable", false).Error + }}, + {name: "detached from the inbound", during: "remove", wantOps: []string{"remove"}, drop: func(db *gorm.DB, inbound *model.Inbound) error { + return db.Where("inbound_id = ?", inbound.Id).Delete(&model.ClientInbound{}).Error + }}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + setupIntegrationDB(t) + db := database.GetDB() + inbound := seedLinkedInboundWithClient(t, "readd-probe-tag", email, 1) + if err := db.Model(&model.ClientRecord{}).Where("email = ?", email).Update("enable", true).Error; err != nil { + t.Fatal(err) + } + if err := db.Create(&xray.ClientTraffic{InboundId: inbound.Id, Email: email, Enable: true}).Error; err != nil { + t.Fatal(err) + } + + core := &fakeCoreAPI{} + core.onOp = func(op string) { + if op == tt.during { + if err := tt.drop(db, inbound); err != nil { + t.Error(err) + } + } + } + startFakeCoreAPI(t, core) + + clients := []model.Client{{Email: email, ID: "11111111-1111-1111-1111-111111111111", Enable: true}} + (&CheckClientIpJob{}).disconnectClientTemporarily(inbound, email, clients) + + core.mu.Lock() + defer core.mu.Unlock() + if !slices.Equal(core.ops, tt.wantOps) { + t.Fatalf("core saw %v, want %v", core.ops, tt.wantOps) + } + }) + } +}