精简版P9(4b): 删除本地核心整条路径(进程管理/buildXrayConfig本地分支/local-only注入/本地核心接口与任务)
- service/xray.go 1741->573行: 只剩 agent 配置构建(GetXrayConfigForAgent/buildAgentXrayConfig=模板+节点入站+egress+转发编译); 删进程生命周期/状态机/RestartXray/StopXray/SetToNeedRestart/GetXrayTraffic/GetOnlineUsers/balancer/TestRoute 与 injectPanelEgress/injectNodeEgresses/injectMtproto/injectAmneziawg/mergeSubscriptionOutbounds
- controller: 43处 SetToNeedRestart 与各 RestartXray 调用清除; 删 balancerStatus/balancerOverride/routeTest/getXrayResult 路由与handler(纯本地核心操作); 前端删 RouteTester 页签与组件
- server service: GetConfigJson 改模板直读; 状态Xray块恒 stopped/空版本; Stop/RestartXrayService 改 no-op(DB导入/恢复流程保留形状)
- setting service: PanelEgressProxyURL/NodeEgressProxyURL 恒直连(桥在本地核心内,已不存在); NodeService.withOutboundBridge 恒直连回退
- runtime: Local 改为全 no-op 桩(DB即真源); NewManager() 去 LocalDeps(APIPort/SetNeedRestart)
- job: 删 XrayTrafficJob/CheckXrayRunningJob/CheckClientIpJob 三文件(共享声明 EventBus/IPWithTimestamp/clientStatsSnapshotMaxClients 迁 shared.go); web.go 不再调度 MtprotoJob/AmneziaWGJob, 删 OnCrash 接线与 stop() 的 sidecar StopAll
- MTProto/AmneziaWG 下线: AddInbound/UpdateInbound 对两协议直接拒绝(存量行保留可删); TG bot 的 restart 命令与在线查询按无本地核心收口
- 测试: GetXrayConfig 系列移植到 agent 构建(finalmask/clients/wireguard/scale); 删退役行为测试(lifecycle/restart/mtproto本地应用/client-ip job/inject系列); 共享测试helper迁 mtproto_test_helpers_test.go 与 job/integration_helpers_test.go
- 既有红(非本轮引入,已在 f6d6520 复现): TestGetXrayConfig_EnabledClientsStillEmitted、TestRouteRegistryContract(P6删 endpoints.ts 遗留)
This commit is contained in:
Muse
+32
-286
@@ -2,313 +2,59 @@ package runtime
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"github.com/mhsanaei/3x-ui/v3/internal/amneziawg"
|
||||
"github.com/mhsanaei/3x-ui/v3/internal/amneziawgnet"
|
||||
"github.com/mhsanaei/3x-ui/v3/internal/database/model"
|
||||
"github.com/mhsanaei/3x-ui/v3/internal/mtproto"
|
||||
"github.com/mhsanaei/3x-ui/v3/internal/xray"
|
||||
)
|
||||
|
||||
type LocalDeps struct {
|
||||
APIPort func() int
|
||||
SetNeedRestart func()
|
||||
}
|
||||
// Local is the runtime for inbounds with no node assignment. The lite
|
||||
// master is a pure control plane — it runs no local xray core and no
|
||||
// MTProto/AmneziaWG sidecars (removed in the P9(4b) round along with the
|
||||
// whole local-core path) — so every operation is a no-op success: the
|
||||
// database is the only state a node-less inbound has, and the service
|
||||
// layer has already written it by the time these are called. Inbounds
|
||||
// only become live once they are homed on an agent node (Remote).
|
||||
type Local struct{}
|
||||
|
||||
type Local struct {
|
||||
deps LocalDeps
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
func NewLocal(deps LocalDeps) *Local {
|
||||
return &Local{deps: deps}
|
||||
}
|
||||
func NewLocal() *Local { return &Local{} }
|
||||
|
||||
func (l *Local) Name() string { return "local" }
|
||||
|
||||
func (l *Local) withAPI(fn func(api *xray.XrayAPI) error) error {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
|
||||
port := l.deps.APIPort()
|
||||
if port <= 0 {
|
||||
return errors.New("local xray is not running")
|
||||
}
|
||||
var api xray.XrayAPI
|
||||
if err := api.Init(port); err != nil {
|
||||
return err
|
||||
}
|
||||
defer api.Close()
|
||||
return fn(&api)
|
||||
}
|
||||
|
||||
func (l *Local) AddInbound(_ context.Context, ib *model.Inbound) error {
|
||||
if ib.Protocol == model.MTProto {
|
||||
inst, ok := mtproto.InstanceFromInbound(ib)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
return mtproto.GetManager().Ensure(inst)
|
||||
}
|
||||
if ib.Protocol == model.AmneziaWG {
|
||||
inst, ok := amneziawg.InstanceFromInbound(ib)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
err := amneziawgnet.GetManager().Ensure(amneziawgnet.Desired{
|
||||
Instance: inst,
|
||||
Options: amneziawgnet.DeviceOptions{
|
||||
HeaderProtectionKey: inst.Obfuscation.HeaderProtectionKey,
|
||||
ContentPaddingAddition: inst.Obfuscation.ContentPaddingAddition,
|
||||
RekeyAfterTime: inst.Obfuscation.RekeyAfterTime,
|
||||
RekeyTimeout: inst.Obfuscation.RekeyTimeout,
|
||||
RejectAfterTime: inst.Obfuscation.RejectAfterTime,
|
||||
KeepaliveTimeout: inst.Obfuscation.KeepaliveTimeout,
|
||||
MaxHandshakeAttempts: inst.Obfuscation.MaxHandshakeAttempts,
|
||||
RandomTrailers: inst.Obfuscation.RandomTrailers,
|
||||
DisableCookies: inst.Obfuscation.DisableCookies,
|
||||
},
|
||||
})
|
||||
// A brand new inbound can be the first one to qualify for
|
||||
// injectAmneziawgnetSocks's Xray-side relay inbound (e.g. its first
|
||||
// valid peer). Ensure only updates the embedded Device -- flag Xray
|
||||
// for a resync so the relay actually gets created within the next
|
||||
// ApplyPendingRestart tick instead of only at the next full restart.
|
||||
if l.deps.SetNeedRestart != nil {
|
||||
l.deps.SetNeedRestart()
|
||||
}
|
||||
return err
|
||||
}
|
||||
body, err := json.MarshalIndent(ib.GenXrayInboundConfig(), "", " ")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return l.withAPI(func(api *xray.XrayAPI) error {
|
||||
return api.AddInbound(body)
|
||||
})
|
||||
}
|
||||
|
||||
func (l *Local) DelInbound(_ context.Context, ib *model.Inbound) error {
|
||||
if ib.Protocol == model.MTProto {
|
||||
mtproto.GetManager().Remove(ib.Id)
|
||||
return nil
|
||||
}
|
||||
if ib.Protocol == model.AmneziaWG {
|
||||
amneziawgnet.GetManager().Remove(ib.Id)
|
||||
// The removed inbound may have been the only one backing Xray's
|
||||
// injectAmneziawgnetSocks relay inbound for this tag -- flag a
|
||||
// resync so the now-stale relay gets torn down promptly.
|
||||
if l.deps.SetNeedRestart != nil {
|
||||
l.deps.SetNeedRestart()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return l.withAPI(func(api *xray.XrayAPI) error {
|
||||
return api.DelInbound(ib.Tag)
|
||||
})
|
||||
}
|
||||
|
||||
func (l *Local) UpdateInbound(ctx context.Context, oldIb, newIb *model.Inbound) error {
|
||||
if oldIb.Protocol == model.MTProto || newIb.Protocol == model.MTProto {
|
||||
return l.updateMtprotoInbound(ctx, oldIb, newIb)
|
||||
}
|
||||
if oldIb.Protocol == model.AmneziaWG || newIb.Protocol == model.AmneziaWG {
|
||||
return l.updateAmneziaWGInbound(ctx, oldIb, newIb)
|
||||
}
|
||||
_ = l.DelInbound(ctx, oldIb)
|
||||
if !newIb.Enable {
|
||||
return nil
|
||||
}
|
||||
return l.AddInbound(ctx, newIb)
|
||||
}
|
||||
|
||||
// updateMtprotoInbound applies an inbound update without the Del+Add sequence
|
||||
// the xray path uses: Remove would drop the manager's fingerprint state, which
|
||||
// is what lets Ensure keep the running mtg process (and its live connections)
|
||||
// when nothing in the generated config changed. The sidecar is only stopped
|
||||
// when the inbound is disabled, loses its last active secret, or moves to a
|
||||
// different protocol.
|
||||
func (l *Local) updateMtprotoInbound(ctx context.Context, oldIb, newIb *model.Inbound) error {
|
||||
if oldIb.Protocol == model.MTProto && newIb.Protocol != model.MTProto {
|
||||
mtproto.GetManager().Remove(oldIb.Id)
|
||||
if !newIb.Enable {
|
||||
return nil
|
||||
}
|
||||
return l.AddInbound(ctx, newIb)
|
||||
}
|
||||
if oldIb.Protocol != model.MTProto {
|
||||
_ = l.DelInbound(ctx, oldIb)
|
||||
}
|
||||
if !newIb.Enable {
|
||||
mtproto.GetManager().Remove(newIb.Id)
|
||||
return nil
|
||||
}
|
||||
inst, ok := mtproto.InstanceFromInbound(newIb)
|
||||
if !ok {
|
||||
mtproto.GetManager().Remove(newIb.Id)
|
||||
return nil
|
||||
}
|
||||
return mtproto.GetManager().Ensure(inst)
|
||||
}
|
||||
|
||||
// updateAmneziaWGInbound mirrors updateMtprotoInbound: it skips the
|
||||
// Remove+Ensure sequence a plain Del+Add would force so that, on an
|
||||
// AmneziaWG-to-AmneziaWG edit, Manager.Ensure's own fingerprint comparison
|
||||
// can reconfigure the running embedded Device in place via IpcSet instead
|
||||
// of always rebuilding it (see internal/amneziawgnet.Manager.ensureLocked --
|
||||
// only an address/MTU change forces a rebuild there, not a peer edit).
|
||||
//
|
||||
// Every exit path below only touches the embedded Device via
|
||||
// amneziawgnet.GetManager() -- none of it rebuilds Xray's own config, which
|
||||
// is what actually creates/removes injectAmneziawgnetSocks's relay inbound.
|
||||
// A peer edit that changes whether this inbound has a qualifying peer at
|
||||
// all (its first peer added, or its last one removed) must still get that
|
||||
// relay created or torn down, so flag Xray for a resync unconditionally
|
||||
// here rather than trying to enumerate which of the branches below need it.
|
||||
func (l *Local) updateAmneziaWGInbound(ctx context.Context, oldIb, newIb *model.Inbound) error {
|
||||
if l.deps.SetNeedRestart != nil {
|
||||
l.deps.SetNeedRestart()
|
||||
}
|
||||
if oldIb.Protocol == model.AmneziaWG && newIb.Protocol != model.AmneziaWG {
|
||||
amneziawgnet.GetManager().Remove(oldIb.Id)
|
||||
if !newIb.Enable {
|
||||
return nil
|
||||
}
|
||||
return l.AddInbound(ctx, newIb)
|
||||
}
|
||||
if oldIb.Protocol != model.AmneziaWG {
|
||||
_ = l.DelInbound(ctx, oldIb)
|
||||
}
|
||||
if !newIb.Enable {
|
||||
amneziawgnet.GetManager().Remove(newIb.Id)
|
||||
return nil
|
||||
}
|
||||
inst, ok := amneziawg.InstanceFromInbound(newIb)
|
||||
if !ok {
|
||||
amneziawgnet.GetManager().Remove(newIb.Id)
|
||||
return nil
|
||||
}
|
||||
return amneziawgnet.GetManager().Ensure(amneziawgnet.Desired{
|
||||
Instance: inst,
|
||||
Options: amneziawgnet.DeviceOptions{
|
||||
HeaderProtectionKey: inst.Obfuscation.HeaderProtectionKey,
|
||||
ContentPaddingAddition: inst.Obfuscation.ContentPaddingAddition,
|
||||
RekeyAfterTime: inst.Obfuscation.RekeyAfterTime,
|
||||
RekeyTimeout: inst.Obfuscation.RekeyTimeout,
|
||||
RejectAfterTime: inst.Obfuscation.RejectAfterTime,
|
||||
KeepaliveTimeout: inst.Obfuscation.KeepaliveTimeout,
|
||||
MaxHandshakeAttempts: inst.Obfuscation.MaxHandshakeAttempts,
|
||||
RandomTrailers: inst.Obfuscation.RandomTrailers,
|
||||
DisableCookies: inst.Obfuscation.DisableCookies,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
func (l *Local) AddUser(_ context.Context, ib *model.Inbound, userMap map[string]any) error {
|
||||
if ib.Protocol == model.MTProto || ib.Protocol == model.AmneziaWG {
|
||||
return nil
|
||||
}
|
||||
return l.withAPI(func(api *xray.XrayAPI) error {
|
||||
return api.AddUser(string(ib.Protocol), ib.Tag, userMap)
|
||||
})
|
||||
}
|
||||
|
||||
func (l *Local) RemoveUser(_ context.Context, ib *model.Inbound, email string) error {
|
||||
if ib.Protocol == model.MTProto || ib.Protocol == model.AmneziaWG {
|
||||
return nil
|
||||
}
|
||||
return l.withAPI(func(api *xray.XrayAPI) error {
|
||||
return api.RemoveUser(ib.Tag, email)
|
||||
})
|
||||
}
|
||||
|
||||
func (l *Local) AddClient(ctx context.Context, ib *model.Inbound, client model.Client) error {
|
||||
if !client.Enable {
|
||||
return nil
|
||||
}
|
||||
user := map[string]any{
|
||||
"email": client.Email,
|
||||
"id": client.ID,
|
||||
"security": client.Security,
|
||||
"flow": client.Flow,
|
||||
"auth": client.Auth,
|
||||
"password": client.Password,
|
||||
"publicKey": client.PublicKey,
|
||||
"allowedIPs": client.AllowedIPs,
|
||||
"preSharedKey": client.PreSharedKey,
|
||||
"keepAlive": wgKeepAlive(client.KeepAlive),
|
||||
}
|
||||
return l.AddUser(ctx, ib, user)
|
||||
}
|
||||
|
||||
func (l *Local) DeleteUser(ctx context.Context, ib *model.Inbound, email string) error {
|
||||
if email == "" {
|
||||
return nil
|
||||
}
|
||||
if err := l.RemoveUser(ctx, ib, email); err != nil {
|
||||
if strings.Contains(err.Error(), "not found") {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
func (l *Local) AddInbound(_ context.Context, _ *model.Inbound) error { return nil }
|
||||
func (l *Local) DelInbound(_ context.Context, _ *model.Inbound) error { return nil }
|
||||
func (l *Local) UpdateInbound(_ context.Context, _, _ *model.Inbound) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (l *Local) DeleteClient(context.Context, string) error {
|
||||
func (l *Local) AddUser(_ context.Context, _ *model.Inbound, _ map[string]any) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (l *Local) UpdateUser(ctx context.Context, ib *model.Inbound, oldEmail string, payload model.Client) error {
|
||||
if oldEmail != "" {
|
||||
if err := l.RemoveUser(ctx, ib, oldEmail); err != nil && !strings.Contains(err.Error(), "not found") {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if !payload.Enable {
|
||||
return nil
|
||||
}
|
||||
user := map[string]any{
|
||||
"email": payload.Email,
|
||||
"id": payload.ID,
|
||||
"security": payload.Security,
|
||||
"flow": payload.Flow,
|
||||
"auth": payload.Auth,
|
||||
"password": payload.Password,
|
||||
"publicKey": payload.PublicKey,
|
||||
"allowedIPs": payload.AllowedIPs,
|
||||
"preSharedKey": payload.PreSharedKey,
|
||||
"keepAlive": wgKeepAlive(payload.KeepAlive),
|
||||
}
|
||||
return l.AddUser(ctx, ib, user)
|
||||
}
|
||||
|
||||
func wgKeepAlive(seconds int) string {
|
||||
if seconds <= 0 {
|
||||
return ""
|
||||
}
|
||||
return strconv.Itoa(seconds)
|
||||
}
|
||||
|
||||
func (l *Local) RestartXray(_ context.Context) error {
|
||||
if l.deps.SetNeedRestart != nil {
|
||||
l.deps.SetNeedRestart()
|
||||
}
|
||||
func (l *Local) RemoveUser(_ context.Context, _ *model.Inbound, _ string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (l *Local) UpdateUser(_ context.Context, _ *model.Inbound, _ string, _ model.Client) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (l *Local) DeleteUser(_ context.Context, _ *model.Inbound, _ string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (l *Local) AddClient(_ context.Context, _ *model.Inbound, _ model.Client) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (l *Local) DeleteClient(_ context.Context, _ string) error { return nil }
|
||||
|
||||
func (l *Local) RestartXray(_ context.Context) error { return nil }
|
||||
|
||||
func (l *Local) ResetClientTraffic(_ context.Context, _ *model.Inbound, _ string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (l *Local) ResetAllTraffics(_ context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (l *Local) ResetInboundTraffic(_ context.Context, _ *model.Inbound) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (l *Local) ResetAllTraffics(_ context.Context) error { return nil }
|
||||
|
||||
@@ -22,9 +22,9 @@ type Manager struct {
|
||||
egressResolver NodeEgressResolver
|
||||
}
|
||||
|
||||
func NewManager(localDeps LocalDeps) *Manager {
|
||||
func NewManager() *Manager {
|
||||
return &Manager{
|
||||
local: NewLocal(localDeps),
|
||||
local: NewLocal(),
|
||||
remotes: make(map[int]*Remote),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,7 +7,7 @@ import (
|
||||
)
|
||||
|
||||
func TestManagerRemoteForRefreshesChangedCredential(t *testing.T) {
|
||||
m := NewManager(LocalDeps{})
|
||||
m := NewManager()
|
||||
first, err := m.RemoteFor(&model.Node{
|
||||
Id: 1,
|
||||
Name: "node",
|
||||
@@ -81,7 +81,7 @@ func TestManagerRemoteForIdentityFields(t *testing.T) {
|
||||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
m := NewManager(LocalDeps{})
|
||||
m := NewManager()
|
||||
firstNode := base
|
||||
first, err := m.RemoteFor(&firstNode)
|
||||
if err != nil {
|
||||
@@ -102,7 +102,7 @@ func TestManagerRemoteForIdentityFields(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestManagerRemoteForClonesInputNode(t *testing.T) {
|
||||
m := NewManager(LocalDeps{})
|
||||
m := NewManager()
|
||||
n := &model.Node{
|
||||
Id: 9,
|
||||
Scheme: "https",
|
||||
|
||||
Reference in New Issue
Block a user