Skip to content

Commit 69265d8

Browse files
x0sinax0sina
committed
feat(controller): add concurrent operation locking and client isolation
- Add `Stopping()` method to Core to check if xray is in stopping state - Improve xray startup error messages to distinguish between crashes and interruptions - Add `syncMu` mutex to Xray to serialize user sync/update operations and prevent race conditions - Add `controlMu` mutex to Controller to serialize control operations across Start/Stop requests - Add `IsCurrentClient()` method to validate that operations come from the controlling client - Add `LockControl()` and `UnlockControl()` methods to Controller for operation serialization - Add `requestClientIP()` helper to extract and validate client IP from requests - Add `validateCurrentClient` middleware to REST API to enforce single-client control - Apply client validation in Start/Stop endpoints to reject requests from non-controlling clients - Apply concurrent operation locking in user sync, update, and restart operations - Prevents concurrent user modifications and ensures atomic state transitions - Ensures only the current controlling client can perform control operations - Fixes potential race conditions and improves multi-client isolation
1 parent 77a0f1b commit 69265d8

10 files changed

Lines changed: 150 additions & 22 deletions

File tree

‎backend/xray/core.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -123,6 +123,12 @@ func (c *Core) Started() bool {
123123
return false
124124
}
125125

126+
func (c *Core) Stopping() bool {
127+
c.mu.Lock()
128+
defer c.mu.Unlock()
129+
return c.stopping
130+
}
131+
126132
func collectUnixSocketPaths(cfg *Config) []string {
127133
if cfg == nil {
128134
return nil

‎backend/xray/jobs.go‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,7 @@ func startupErrorWithTail(core *Core, tailSize int, reason string) error {
3737
return errors.New(reason)
3838
}
3939

40-
return fmt.Errorf("%s; recent xray logs:\n%s", reason, strings.Join(tail, "\n"))
40+
return fmt.Errorf("%s; no fatal xray startup log was detected. Recent xray logs:\n%s", reason, strings.Join(tail, "\n"))
4141
}
4242

4343
func (x *Xray) checkXrayStatus(baseCtx context.Context) error {
@@ -72,7 +72,11 @@ func (x *Xray) checkXrayStatus(baseCtx context.Context) error {
7272

7373
// No error in logs, check API
7474
if !x.core.Started() {
75-
return startupErrorWithTail(x.core, x.startupLogTailSize(), "xray process stopped before API became ready")
75+
reason := "xray process stopped before API became ready"
76+
if x.core.Stopping() {
77+
reason = "xray startup was interrupted by a stop/restart request before API became ready"
78+
}
79+
return startupErrorWithTail(x.core, x.startupLogTailSize(), reason)
7680
}
7781
}
7882
}

‎backend/xray/user.go‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -120,6 +120,9 @@ func isActiveInbound(inbound *Inbound, inbounds []string, settings api.ProxySett
120120
}
121121

122122
func (x *Xray) SyncUser(ctx context.Context, user *common.User) error {
123+
x.syncMu.Lock()
124+
defer x.syncMu.Unlock()
125+
123126
proxySetting, err := setupUserAccount(user)
124127
if err != nil {
125128
return err
@@ -158,6 +161,9 @@ func (x *Xray) SyncUser(ctx context.Context, user *common.User) error {
158161
}
159162

160163
func (x *Xray) SyncUsers(ctx context.Context, users []*common.User) error {
164+
x.syncMu.Lock()
165+
defer x.syncMu.Unlock()
166+
161167
candidate, err := x.config.Clone()
162168
if err != nil {
163169
return err
@@ -209,6 +215,9 @@ func (x *Xray) restorePreviousConfig(previous *Config) error {
209215
}
210216

211217
func (x *Xray) UpdateUsers(ctx context.Context, users []*common.User) error {
218+
x.syncMu.Lock()
219+
defer x.syncMu.Unlock()
220+
212221
handler := x.handler
213222
inboundByTag, updates := x.config.buildInboundUpdates(users)
214223
var errMessage string
@@ -243,6 +252,9 @@ func (x *Xray) UpdateUsers(ctx context.Context, users []*common.User) error {
243252
}
244253

245254
func (x *Xray) UpdateUsersAndRestart(ctx context.Context, users []*common.User) error {
255+
x.syncMu.Lock()
256+
defer x.syncMu.Unlock()
257+
246258
candidate, err := x.config.Clone()
247259
if err != nil {
248260
return err

‎backend/xray/xray.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ type Xray struct {
2020
metricPort int
2121
cancelFunc context.CancelFunc
2222
mu sync.RWMutex
23+
syncMu sync.Mutex
2324
}
2425

2526
func New(ctx context.Context, xrayConfig *Config, users []*common.User, apiPort, metricPort int, cfg *config.Config) (*Xray, error) {

‎controller/controller.go‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@ type Controller struct {
3434
stats *common.SystemStatsResponse
3535
cancelFunc context.CancelFunc
3636
mu sync.RWMutex
37+
controlMu sync.Mutex
3738
}
3839

3940
func New(cfg *config.Config) *Controller {
@@ -94,6 +95,20 @@ func (c *Controller) Ip() string {
9495
return c.clientIP
9596
}
9697

98+
func (c *Controller) IsCurrentClient(ip string) bool {
99+
c.mu.RLock()
100+
defer c.mu.RUnlock()
101+
return c.clientIP == "" || c.clientIP == ip
102+
}
103+
104+
func (c *Controller) LockControl() {
105+
c.controlMu.Lock()
106+
}
107+
108+
func (c *Controller) UnlockControl() {
109+
c.controlMu.Unlock()
110+
}
111+
97112
func (c *Controller) NewRequest() {
98113
c.mu.Lock()
99114
defer c.mu.Unlock()

‎controller/rest/base.go‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@ package rest
22

33
import (
44
"log"
5-
"net"
65
"net/http"
76

87
"github.com/pasarguard/node/common"
@@ -13,20 +12,27 @@ func (s *Service) Base(w http.ResponseWriter, _ *http.Request) {
1312
}
1413

1514
func (s *Service) Start(w http.ResponseWriter, r *http.Request) {
15+
s.LockControl()
16+
defer s.UnlockControl()
17+
1618
data := &common.Backend{}
1719

1820
if err := common.ReadProtoBody(r.Body, data); err != nil {
1921
http.Error(w, err.Error(), http.StatusBadRequest)
2022
return
2123
}
2224

23-
ip, _, err := net.SplitHostPort(r.RemoteAddr)
24-
if err != nil {
25+
ip, ok := requestClientIP(r)
26+
if !ok {
2527
http.Error(w, "unknown ip", http.StatusServiceUnavailable)
2628
return
2729
}
2830

2931
if s.Backend() != nil {
32+
if !s.IsCurrentClient(ip) {
33+
http.Error(w, "node is controlled by another client", http.StatusForbidden)
34+
return
35+
}
3036
log.Println("New connection from ", ip, " core control access was taken away from previous client.")
3137
s.Disconnect()
3238
}
@@ -42,6 +48,9 @@ func (s *Service) Start(w http.ResponseWriter, r *http.Request) {
4248
}
4349

4450
func (s *Service) Stop(w http.ResponseWriter, _ *http.Request) {
51+
s.LockControl()
52+
defer s.UnlockControl()
53+
4554
s.Disconnect()
4655

4756
common.SendProtoResponse(w, &common.Empty{})

‎controller/rest/middleware.go‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ package rest
22

33
import (
44
"log"
5+
"net"
56
"net/http"
67

78
"github.com/go-chi/chi/v5/middleware"
@@ -33,6 +34,30 @@ func (s *Service) validateApiKey(next http.Handler) http.Handler {
3334
})
3435
}
3536

37+
func requestClientIP(r *http.Request) (string, bool) {
38+
ip, _, err := net.SplitHostPort(r.RemoteAddr)
39+
if err != nil {
40+
return "", false
41+
}
42+
return ip, true
43+
}
44+
45+
func (s *Service) validateCurrentClient(next http.Handler) http.Handler {
46+
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
47+
ip, ok := requestClientIP(r)
48+
if !ok {
49+
http.Error(w, "unknown ip", http.StatusServiceUnavailable)
50+
return
51+
}
52+
if !s.IsCurrentClient(ip) {
53+
http.Error(w, "node is controlled by another client", http.StatusForbidden)
54+
return
55+
}
56+
57+
next.ServeHTTP(w, r)
58+
})
59+
}
60+
3661
func (s *Service) checkBackendMiddleware(next http.Handler) http.Handler {
3762
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
3863
back := s.Backend()

‎controller/rest/service.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@ func (s *Service) setRouter() {
3636

3737
router.Group(func(private chi.Router) {
3838
private.Use(s.checkBackendMiddleware)
39+
private.Use(s.validateCurrentClient)
3940

4041
private.Put("/stop", s.Stop)
4142
private.Get("/logs", s.GetLogs)

‎controller/rpc/base.go‎

Lines changed: 14 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -3,31 +3,25 @@ package rpc
33
import (
44
"context"
55
"log"
6-
"net"
76

87
"github.com/pasarguard/node/common"
9-
"google.golang.org/grpc/peer"
8+
"google.golang.org/grpc/codes"
9+
"google.golang.org/grpc/status"
1010
)
1111

1212
func (s *Service) Start(ctx context.Context, data *common.Backend) (*common.BaseInfoResponse, error) {
13-
clientIP := ""
14-
if p, ok := peer.FromContext(ctx); ok {
15-
// Extract IP address from peer address
16-
if tcpAddr, ok := p.Addr.(*net.TCPAddr); ok {
17-
clientIP = tcpAddr.IP.String()
18-
} else {
19-
// For other address types, extract just the IP without the port
20-
addr := p.Addr.String()
21-
if host, _, err := net.SplitHostPort(addr); err == nil {
22-
clientIP = host
23-
} else {
24-
// If SplitHostPort fails, use the whole address
25-
clientIP = addr
26-
}
27-
}
13+
s.LockControl()
14+
defer s.UnlockControl()
15+
16+
clientIP := clientIPFromContext(ctx)
17+
if clientIP == "" {
18+
return nil, status.Errorf(codes.PermissionDenied, "unknown client ip")
2819
}
2920

3021
if s.Backend() != nil {
22+
if !s.IsCurrentClient(clientIP) {
23+
return nil, status.Errorf(codes.PermissionDenied, "node is controlled by another client")
24+
}
3125
log.Println("New connection from ", clientIP, " core control access was taken away from previous client.")
3226
s.Disconnect()
3327
}
@@ -42,6 +36,9 @@ func (s *Service) Start(ctx context.Context, data *common.Backend) (*common.Base
4236
}
4337

4438
func (s *Service) Stop(_ context.Context, _ *common.Empty) (*common.Empty, error) {
39+
s.LockControl()
40+
defer s.UnlockControl()
41+
4542
s.Disconnect()
4643
return nil, nil
4744
}

‎controller/rpc/middleware.go‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"context"
55
"fmt"
66
"log"
7+
"net"
78
"strings"
89

910
"github.com/google/uuid"
@@ -15,6 +16,20 @@ import (
1516
"google.golang.org/grpc/status"
1617
)
1718

19+
func clientIPFromContext(ctx context.Context) string {
20+
if p, ok := peer.FromContext(ctx); ok {
21+
if tcpAddr, ok := p.Addr.(*net.TCPAddr); ok {
22+
return tcpAddr.IP.String()
23+
}
24+
addr := p.Addr.String()
25+
if host, _, err := net.SplitHostPort(addr); err == nil {
26+
return host
27+
}
28+
return addr
29+
}
30+
return ""
31+
}
32+
1833
func validateApiKey(ctx context.Context, s *Service) error {
1934
// Extract metadata
2035
md, ok := metadata.FromIncomingContext(ctx)
@@ -58,6 +73,47 @@ func validateApiKeyMiddleware(s *Service) grpc.UnaryServerInterceptor {
5873
}
5974
}
6075

76+
func validateCurrentClient(ctx context.Context, s *Service) error {
77+
clientIP := clientIPFromContext(ctx)
78+
if clientIP == "" {
79+
return status.Errorf(codes.PermissionDenied, "unknown client ip")
80+
}
81+
if !s.IsCurrentClient(clientIP) {
82+
return status.Errorf(codes.PermissionDenied, "node is controlled by another client")
83+
}
84+
return nil
85+
}
86+
87+
func validateCurrentClientMiddleware(s *Service) grpc.UnaryServerInterceptor {
88+
return func(
89+
ctx context.Context,
90+
req interface{},
91+
info *grpc.UnaryServerInfo,
92+
handler grpc.UnaryHandler,
93+
) (interface{}, error) {
94+
if err := validateCurrentClient(ctx, s); err != nil {
95+
return nil, err
96+
}
97+
98+
return handler(ctx, req)
99+
}
100+
}
101+
102+
func validateCurrentClientStreamMiddleware(s *Service) grpc.StreamServerInterceptor {
103+
return func(
104+
srv interface{},
105+
ss grpc.ServerStream,
106+
info *grpc.StreamServerInfo,
107+
handler grpc.StreamHandler,
108+
) error {
109+
if err := validateCurrentClient(ss.Context(), s); err != nil {
110+
return err
111+
}
112+
113+
return handler(srv, ss)
114+
}
115+
}
116+
61117
func validateApiKeyStreamMiddleware(s *Service) grpc.StreamServerInterceptor {
62118
return func(
63119
srv interface{},
@@ -211,6 +267,7 @@ func ConditionalMiddleware(s *Service) grpc.UnaryServerInterceptor {
211267
interceptors = append(interceptors, validateApiKeyMiddleware(s))
212268

213269
if backendMethods[info.FullMethod] {
270+
interceptors = append(interceptors, validateCurrentClientMiddleware(s))
214271
interceptors = append(interceptors, CheckBackendMiddleware(s))
215272
}
216273

@@ -233,6 +290,7 @@ func ConditionalStreamMiddleware(s *Service) grpc.StreamServerInterceptor {
233290
interceptors = append(interceptors, validateApiKeyStreamMiddleware(s))
234291

235292
if backendMethods[info.FullMethod] {
293+
interceptors = append(interceptors, validateCurrentClientStreamMiddleware(s))
236294
interceptors = append(interceptors, CheckBackendStreamMiddleware(s))
237295
}
238296

0 commit comments

Comments
 (0)