mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-11-02 11:15:03 +03:00
fix(limit-ip): do not re-add a client the database stopped serving
disconnectClientTemporarily removed an over-limit client from the core and re-added the snapshot the scan had loaded 100 ms later, so a disable or detach committed in between was undone: the traffic job's own RemoveUser had already found nothing, and the disabled account stayed live in Xray until the next restart. The re-add now rereads the client and its inbound and skips one that is disabled, expired, over quota or no longer attached.
This commit is contained in:
@@ -820,11 +820,43 @@ func (j *CheckClientIpJob) disconnectClientTemporarily(inbound *model.Inbound, c
|
|||||||
// only widens the window in which new handshakes fail.
|
// only widens the window in which new handshakes fail.
|
||||||
time.Sleep(100 * time.Millisecond)
|
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
|
// Re-add user to allow new connections
|
||||||
err = xrayAPI.AddUser(protocol, inbound.Tag, clientConfig)
|
err = xrayAPI.AddUser(protocol, inbound.Tag, clientConfig)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logger.Warningf("[LIMIT_IP] Failed to re-add user %s: %v", clientEmail, err)
|
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.
|
// resolveXrayAPIPort returns the API inbound port from running config, then template config, then default.
|
||||||
|
|||||||
@@ -1,11 +1,24 @@
|
|||||||
package job
|
package job
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"net"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"slices"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"testing"
|
"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/database/model"
|
||||||
"github.com/mhsanaei/3x-ui/v3/internal/logger"
|
"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
|
// 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")
|
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)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user