From df9665b99052092adfa777cdb6ec65640951c091 Mon Sep 17 00:00:00 2001 From: Rerowros Date: Mon, 10 Aug 2026 08:51:33 +0400 Subject: [PATCH 1/2] fix(node): harden runtime and fence user sync epochs --- backend/xray/user.go | 73 ++++- backend/xray/user_removal_test.go | 32 +++ backend/xray/user_runtime_consistency_test.go | 258 ++++++++++++++++++ backend/xray/xray.go | 20 +- common/helper.go | 12 +- common/service.pb.go | 80 +++++- common/service.proto | 6 + config/config.go | 62 +++-- config/config_test.go | 37 +++ controller/controller.go | 168 +++++++++--- controller/controller_lifecycle_test.go | 92 +++++++ controller/controller_test.go | 31 +++ controller/rest/base.go | 24 +- controller/rest/http_timeout.go | 76 ++++++ controller/rest/log.go | 18 +- controller/rest/log_timeout_test.go | 185 +++++++++++++ controller/rest/middleware.go | 4 + controller/rest/service.go | 8 +- controller/rest/user.go | 82 ++++-- controller/rest/user_sync_epoch.go | 17 ++ controller/rest/user_sync_epoch_test.go | 66 +++++ controller/rpc/base.go | 19 +- controller/rpc/middleware.go | 4 +- controller/rpc/user.go | 100 +++++-- controller/rpc/user_sync_epoch.go | 17 ++ controller/rpc/user_sync_epoch_test.go | 47 ++++ controller/user_sync_epoch_test.go | 203 ++++++++++++++ docker-compose.yml | 5 +- 28 files changed, 1568 insertions(+), 178 deletions(-) create mode 100644 backend/xray/user_removal_test.go create mode 100644 backend/xray/user_runtime_consistency_test.go create mode 100644 config/config_test.go create mode 100644 controller/controller_lifecycle_test.go create mode 100644 controller/controller_test.go create mode 100644 controller/rest/http_timeout.go create mode 100644 controller/rest/log_timeout_test.go create mode 100644 controller/rest/user_sync_epoch.go create mode 100644 controller/rest/user_sync_epoch_test.go create mode 100644 controller/rpc/user_sync_epoch.go create mode 100644 controller/rpc/user_sync_epoch_test.go create mode 100644 controller/user_sync_epoch_test.go diff --git a/backend/xray/user.go b/backend/xray/user.go index 671622b..26ed610 100644 --- a/backend/xray/user.go +++ b/backend/xray/user.go @@ -6,13 +6,28 @@ import ( "fmt" "log" "slices" + "sort" "strings" "time" "github.com/pasarguard/node/backend/xray/api" "github.com/pasarguard/node/common" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) +type inboundUserHandler interface { + AddInboundUser(context.Context, string, api.Account) error + RemoveInboundUser(context.Context, string, string) error +} + +func (x *Xray) inboundUserHandler() inboundUserHandler { + if x.userHandler != nil { + return x.userHandler + } + return x.handler +} + func setupUserAccount(user *common.User) (api.ProxySettings, error) { settings := api.ProxySettings{} if user.GetProxies().GetVmess() != nil { @@ -43,6 +58,32 @@ func setupUserAccount(user *common.User) (api.ProxySettings, error) { return settings, nil } +// removeInboundUser treats an already-absent runtime user as a successful +// idempotent revoke, but never hides transport/core errors. A caller must not +// report a revoked credential while Xray still accepts it. +func removeInboundUser(ctx context.Context, handler inboundUserHandler, tag, email string) error { + err := handler.RemoveInboundUser(ctx, tag, email) + if isBenignUserRemovalError(err) { + return nil + } + return err +} + +func isBenignUserRemovalError(err error) bool { + if err == nil || status.Code(err) == codes.NotFound { + return true + } + // Xray's HandlerService historically reports both of these idempotent + // conditions as Unknown rather than NotFound. An absent user cannot keep a + // credential alive, and API/non-user-manager inbounds cannot contain one. + // All transport and runtime failures stay visible to the caller. + if status.Code(err) != codes.Unknown { + return false + } + message := strings.ToLower(status.Convert(err).Message()) + return strings.Contains(message, "not found") || strings.Contains(message, "not a usermanager") +} + func inboundFlow(inbound *Inbound) string { if inbound == nil || inbound.Settings == nil { return "" @@ -128,7 +169,7 @@ func (x *Xray) SyncUser(ctx context.Context, user *common.User) error { return err } - handler := x.handler + handler := x.inboundUserHandler() inbounds := x.config.InboundConfigs var errMessage strings.Builder @@ -140,17 +181,22 @@ func (x *Xray) SyncUser(ctx context.Context, user *common.User) error { continue } - _ = handler.RemoveInboundUser(ctx, inbound.Tag, user.Email) + if err := removeInboundUser(ctx, handler, inbound.Tag, user.Email); err != nil { + return fmt.Errorf("failed to remove user %q from inbound %q: %w", user.Email, inbound.Tag, err) + } + // Keep the restart snapshot aligned with every confirmed runtime + // mutation. In particular, a failed replacement must not leave the old + // credential in the snapshot where a health restart could resurrect it. + inbound.removeUser(user.Email) account, isActive := isActiveInbound(inbound, userInbounds, proxySetting) if isActive { - inbound.updateUser(account) err = handler.AddInboundUser(ctx, inbound.Tag, accountForAPI(inbound, account)) if err != nil { log.Println(err) errMessage.WriteString("\n" + err.Error()) + } else { + inbound.updateUser(account) } - } else { - inbound.removeUser(user.GetEmail()) } } @@ -218,7 +264,7 @@ func (x *Xray) UpdateUsers(ctx context.Context, users []*common.User) error { x.syncMu.Lock() defer x.syncMu.Unlock() - handler := x.handler + handler := x.inboundUserHandler() inboundByTag, updates := x.config.buildInboundUpdates(users) var errMessage strings.Builder @@ -227,20 +273,29 @@ func (x *Xray) UpdateUsers(ctx context.Context, users []*common.User) error { for email := range update.removeEmailSet { removeEmails = append(removeEmails, email) } + sort.Strings(removeEmails) inbound := inboundByTag[tag] - inbound.updateUsers(update.accounts, removeEmails) for _, email := range removeEmails { - handler.RemoveInboundUser(ctx, tag, email) + if err := removeInboundUser(ctx, handler, tag, email); err != nil { + return fmt.Errorf("failed to remove user %q from inbound %q: %w", email, tag, err) + } + inbound.removeUser(email) } for _, account := range update.accounts { - _ = handler.RemoveInboundUser(ctx, tag, account.GetEmail()) + email := account.GetEmail() + if err := removeInboundUser(ctx, handler, tag, email); err != nil { + return fmt.Errorf("failed to replace user %q in inbound %q: %w", email, tag, err) + } + inbound.removeUser(email) if err := handler.AddInboundUser(ctx, tag, accountForAPI(inbound, account)); err != nil { log.Println(err) errMessage.WriteString("\n" + err.Error()) + continue } + inbound.updateUser(account) } } diff --git a/backend/xray/user_removal_test.go b/backend/xray/user_removal_test.go new file mode 100644 index 0000000..0ed3dac --- /dev/null +++ b/backend/xray/user_removal_test.go @@ -0,0 +1,32 @@ +package xray + +import ( + "errors" + "testing" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func TestIsBenignUserRemovalError(t *testing.T) { + for _, err := range []error{ + nil, + status.Error(codes.NotFound, "user not found"), + status.Error(codes.Unknown, "proxy/trojan: User user@example.com not found."), + status.Error(codes.Unknown, "app/proxyman/command: proxy is not a UserManager"), + } { + if !isBenignUserRemovalError(err) { + t.Fatalf("expected benign removal error: %v", err) + } + } + + for _, err := range []error{ + status.Error(codes.Unavailable, "connection refused"), + status.Error(codes.DeadlineExceeded, "deadline exceeded"), + errors.New("local handler failure"), + } { + if isBenignUserRemovalError(err) { + t.Fatalf("unexpectedly accepted runtime failure: %v", err) + } + } +} diff --git a/backend/xray/user_runtime_consistency_test.go b/backend/xray/user_runtime_consistency_test.go new file mode 100644 index 0000000..800f068 --- /dev/null +++ b/backend/xray/user_runtime_consistency_test.go @@ -0,0 +1,258 @@ +package xray + +import ( + "context" + "encoding/json" + "errors" + "testing" + + "github.com/pasarguard/node/backend/xray/api" + "github.com/pasarguard/node/common" + "github.com/xtls/xray-core/infra/conf" +) + +type scriptedInboundUserHandler struct { + runtime map[string]api.Account + removeCalls int + addCalls int + removeFailures map[int]error + addFailures map[int]error +} + +func (h *scriptedInboundUserHandler) RemoveInboundUser(_ context.Context, tag, email string) error { + h.removeCalls++ + if err := h.removeFailures[h.removeCalls]; err != nil { + return err + } + delete(h.runtime, tag+"\x00"+email) + return nil +} + +func (h *scriptedInboundUserHandler) AddInboundUser(_ context.Context, tag string, account api.Account) error { + h.addCalls++ + if err := h.addFailures[h.addCalls]; err != nil { + return err + } + h.runtime[tag+"\x00"+account.GetEmail()] = account + return nil +} + +func trojanAccount(email, password string) *api.TrojanAccount { + return &api.TrojanAccount{ + BaseAccount: api.BaseAccount{Email: email}, + Password: password, + } +} + +func trojanUser(email, password string, inbounds ...string) *common.User { + return &common.User{ + Email: email, + Inbounds: inbounds, + Proxies: &common.Proxy{ + Trojan: &common.Trojan{Password: password}, + }, + } +} + +func newRuntimeConsistencyXray(accounts ...*api.TrojanAccount) (*Xray, *Inbound, *scriptedInboundUserHandler) { + const tag = "trojan-in" + clients := make(map[string]api.Account, len(accounts)) + runtime := make(map[string]api.Account, len(accounts)) + for _, account := range accounts { + clients[account.GetEmail()] = account + runtime[tag+"\x00"+account.GetEmail()] = account + } + inbound := &Inbound{ + Tag: tag, + Protocol: Trojan, + Settings: make(map[string]any), + clients: clients, + } + handler := &scriptedInboundUserHandler{ + runtime: runtime, + removeFailures: make(map[int]error), + addFailures: make(map[int]error), + } + x := &Xray{ + config: &Config{ + LogConfig: &conf.LogConfig{}, + InboundConfigs: []*Inbound{inbound}, + }, + userHandler: handler, + } + return x, inbound, handler +} + +func accountPassword(t *testing.T, accounts map[string]api.Account, email string) (string, bool) { + t.Helper() + account, ok := accounts[email] + if !ok { + return "", false + } + trojan, ok := account.(*api.TrojanAccount) + if !ok { + t.Fatalf("unexpected account type for %q: %T", email, account) + } + return trojan.Password, true +} + +func runtimePassword(t *testing.T, handler *scriptedInboundUserHandler, email string) (string, bool) { + t.Helper() + return accountPassword(t, handler.runtime, "trojan-in\x00"+email) +} + +func restartSnapshotPasswords(t *testing.T, config *Config) map[string]string { + t.Helper() + payload, err := config.ToBytes() + if err != nil { + t.Fatal(err) + } + var snapshot struct { + Inbounds []struct { + Settings struct { + Clients []struct { + Email string `json:"email"` + Password string `json:"password"` + } `json:"clients"` + } `json:"settings"` + } `json:"inbounds"` + } + if err := json.Unmarshal(payload, &snapshot); err != nil { + t.Fatal(err) + } + passwords := make(map[string]string) + for _, inbound := range snapshot.Inbounds { + for _, client := range inbound.Settings.Clients { + passwords[client.Email] = client.Password + } + } + return passwords +} + +func TestUpdateUsersPartialAddFailureKeepsRuntimeAndRestartSnapshotAligned(t *testing.T) { + oldA := trojanAccount("a@example.com", "old-a") + oldB := trojanAccount("b@example.com", "old-b") + x, inbound, handler := newRuntimeConsistencyXray(oldA, oldB) + handler.addFailures[2] = errors.New("second add failed") + + err := x.UpdateUsers(context.Background(), []*common.User{ + trojanUser("a@example.com", "new-a", inbound.Tag), + trojanUser("b@example.com", "new-b", inbound.Tag), + }) + if err == nil { + t.Fatal("expected partial add failure") + } + + if password, ok := accountPassword(t, inbound.clients, "a@example.com"); !ok || password != "new-a" { + t.Fatalf("cached successful replacement = %q, %v; want new-a, true", password, ok) + } + if _, ok := accountPassword(t, inbound.clients, "b@example.com"); ok { + t.Fatal("failed replacement retained the old cached credential") + } + if password, ok := runtimePassword(t, handler, "a@example.com"); !ok || password != "new-a" { + t.Fatalf("runtime successful replacement = %q, %v; want new-a, true", password, ok) + } + if _, ok := runtimePassword(t, handler, "b@example.com"); ok { + t.Fatal("failed replacement remained in runtime") + } + + snapshot := restartSnapshotPasswords(t, x.config) + if snapshot["a@example.com"] != "new-a" { + t.Fatalf("restart snapshot lost successful replacement: %#v", snapshot) + } + if _, ok := snapshot["b@example.com"]; ok { + t.Fatalf("restart snapshot would resurrect failed replacement: %#v", snapshot) + } +} + +func TestUpdateUsersPartialRemoveFailureKeepsConfirmedRemovalInSnapshot(t *testing.T) { + oldA := trojanAccount("a@example.com", "old-a") + oldB := trojanAccount("b@example.com", "old-b") + x, inbound, handler := newRuntimeConsistencyXray(oldA, oldB) + handler.removeFailures[2] = errors.New("second remove failed") + + err := x.UpdateUsers(context.Background(), []*common.User{ + trojanUser("a@example.com", "unused"), + trojanUser("b@example.com", "unused"), + }) + if err == nil { + t.Fatal("expected partial remove failure") + } + + if _, ok := accountPassword(t, inbound.clients, "a@example.com"); ok { + t.Fatal("confirmed removal remained in cache") + } + if password, ok := accountPassword(t, inbound.clients, "b@example.com"); !ok || password != "old-b" { + t.Fatalf("failed removal changed cached credential = %q, %v", password, ok) + } + if _, ok := runtimePassword(t, handler, "a@example.com"); ok { + t.Fatal("confirmed removal remained in runtime") + } + if password, ok := runtimePassword(t, handler, "b@example.com"); !ok || password != "old-b" { + t.Fatalf("failed removal changed runtime credential = %q, %v", password, ok) + } + + snapshot := restartSnapshotPasswords(t, x.config) + if _, ok := snapshot["a@example.com"]; ok { + t.Fatalf("restart snapshot would resurrect confirmed removal: %#v", snapshot) + } + if snapshot["b@example.com"] != "old-b" { + t.Fatalf("restart snapshot lost failed removal credential: %#v", snapshot) + } +} + +func TestUpdateUsersContinuesAfterAddFailureAndCommitsLaterSuccess(t *testing.T) { + oldA := trojanAccount("a@example.com", "old-a") + oldB := trojanAccount("b@example.com", "old-b") + x, inbound, handler := newRuntimeConsistencyXray(oldA, oldB) + handler.addFailures[1] = errors.New("first add failed") + + err := x.UpdateUsers(context.Background(), []*common.User{ + trojanUser("a@example.com", "new-a", inbound.Tag), + trojanUser("b@example.com", "new-b", inbound.Tag), + }) + if err == nil { + t.Fatal("expected partial add failure") + } + + if _, ok := accountPassword(t, inbound.clients, "a@example.com"); ok { + t.Fatal("failed replacement retained the old cached credential") + } + if password, ok := accountPassword(t, inbound.clients, "b@example.com"); !ok || password != "new-b" { + t.Fatalf("later successful replacement = %q, %v; want new-b, true", password, ok) + } + if _, ok := runtimePassword(t, handler, "a@example.com"); ok { + t.Fatal("failed replacement remained in runtime") + } + if password, ok := runtimePassword(t, handler, "b@example.com"); !ok || password != "new-b" { + t.Fatalf("later successful runtime replacement = %q, %v; want new-b, true", password, ok) + } + + snapshot := restartSnapshotPasswords(t, x.config) + if _, ok := snapshot["a@example.com"]; ok { + t.Fatalf("restart snapshot would resurrect failed replacement: %#v", snapshot) + } + if snapshot["b@example.com"] != "new-b" { + t.Fatalf("restart snapshot lost later successful replacement: %#v", snapshot) + } +} + +func TestSyncUserAddFailureDoesNotResurrectRemovedCredential(t *testing.T) { + old := trojanAccount("user@example.com", "old") + x, inbound, handler := newRuntimeConsistencyXray(old) + handler.addFailures[1] = errors.New("add failed") + + err := x.SyncUser(context.Background(), trojanUser("user@example.com", "new", inbound.Tag)) + if err == nil { + t.Fatal("expected add failure") + } + if _, ok := accountPassword(t, inbound.clients, "user@example.com"); ok { + t.Fatal("failed replacement retained the old cached credential") + } + if _, ok := runtimePassword(t, handler, "user@example.com"); ok { + t.Fatal("failed replacement remained in runtime") + } + if snapshot := restartSnapshotPasswords(t, x.config); len(snapshot) != 0 { + t.Fatalf("restart snapshot would resurrect removed credential: %#v", snapshot) + } +} diff --git a/backend/xray/xray.go b/backend/xray/xray.go index 205a611..981f6df 100644 --- a/backend/xray/xray.go +++ b/backend/xray/xray.go @@ -13,14 +13,15 @@ import ( ) type Xray struct { - config *Config - cfg *config.Config - core *Core - handler *api.XrayHandler - metricPort int - cancelFunc context.CancelFunc - mu sync.RWMutex - syncMu sync.Mutex + config *Config + cfg *config.Config + core *Core + handler *api.XrayHandler + userHandler inboundUserHandler + metricPort int + cancelFunc context.CancelFunc + mu sync.RWMutex + syncMu sync.Mutex } func New(ctx context.Context, xrayConfig *Config, users []*common.User, apiPort, metricPort int, cfg *config.Config) (*Xray, error) { @@ -123,6 +124,9 @@ func (x *Xray) Started() bool { } func (x *Xray) Restart() error { + x.syncMu.Lock() + defer x.syncMu.Unlock() + return x.restartCoreWithConfig(x.config) } diff --git a/common/helper.go b/common/helper.go index 54a0da5..3c0e010 100644 --- a/common/helper.go +++ b/common/helper.go @@ -3,6 +3,7 @@ package common import ( "crypto/sha256" "encoding/base64" + "errors" "io" "net/http" "strings" @@ -12,12 +13,19 @@ import ( "google.golang.org/protobuf/proto" ) +const MaxProtoBodyBytes int64 = 64 * 1024 * 1024 + +var ErrProtoBodyTooLarge = errors.New("protobuf request body exceeds the maximum size") + func ReadProtoBody(body io.ReadCloser, message proto.Message) error { - data, err := io.ReadAll(body) + defer body.Close() + data, err := io.ReadAll(io.LimitReader(body, MaxProtoBodyBytes+1)) if err != nil { return err } - defer body.Close() + if int64(len(data)) > MaxProtoBodyBytes { + return ErrProtoBodyTooLarge + } // Decode into a map if err = proto.Unmarshal(data, message); err != nil { diff --git a/common/service.pb.go b/common/service.pb.go index a4b9a8d..ef1f3c8 100644 --- a/common/service.pb.go +++ b/common/service.pb.go @@ -163,12 +163,14 @@ func (*Empty) Descriptor() ([]byte, []int) { // Base info response message type BaseInfoResponse struct { - state protoimpl.MessageState `protogen:"open.v1"` - Started bool `protobuf:"varint,1,opt,name=started,proto3" json:"started,omitempty"` - CoreVersion string `protobuf:"bytes,2,opt,name=core_version,json=coreVersion,proto3" json:"core_version,omitempty"` - NodeVersion string `protobuf:"bytes,3,opt,name=node_version,json=nodeVersion,proto3" json:"node_version,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + Started bool `protobuf:"varint,1,opt,name=started,proto3" json:"started,omitempty"` + CoreVersion string `protobuf:"bytes,2,opt,name=core_version,json=coreVersion,proto3" json:"core_version,omitempty"` + NodeVersion string `protobuf:"bytes,3,opt,name=node_version,json=nodeVersion,proto3" json:"node_version,omitempty"` + UserSyncEpochSupported bool `protobuf:"varint,4,opt,name=user_sync_epoch_supported,json=userSyncEpochSupported,proto3" json:"user_sync_epoch_supported,omitempty"` + UserSyncEpoch uint64 `protobuf:"varint,5,opt,name=user_sync_epoch,json=userSyncEpoch,proto3" json:"user_sync_epoch,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *BaseInfoResponse) Reset() { @@ -222,6 +224,20 @@ func (x *BaseInfoResponse) GetNodeVersion() string { return "" } +func (x *BaseInfoResponse) GetUserSyncEpochSupported() bool { + if x != nil { + return x.UserSyncEpochSupported + } + return false +} + +func (x *BaseInfoResponse) GetUserSyncEpoch() uint64 { + if x != nil { + return x.UserSyncEpoch + } + return 0 +} + type Backend struct { state protoimpl.MessageState `protogen:"open.v1"` Type BackendType `protobuf:"varint,1,opt,name=type,proto3,enum=service.BackendType" json:"type,omitempty"` @@ -229,6 +245,7 @@ type Backend struct { Users []*User `protobuf:"bytes,3,rep,name=users,proto3" json:"users,omitempty"` KeepAlive uint64 `protobuf:"varint,4,opt,name=keep_alive,json=keepAlive,proto3" json:"keep_alive,omitempty"` ExcludeInbounds []string `protobuf:"bytes,5,rep,name=exclude_inbounds,json=excludeInbounds,proto3" json:"exclude_inbounds,omitempty"` + UserSyncEpoch uint64 `protobuf:"varint,6,opt,name=user_sync_epoch,json=userSyncEpoch,proto3" json:"user_sync_epoch,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -298,6 +315,13 @@ func (x *Backend) GetExcludeInbounds() []string { return nil } +func (x *Backend) GetUserSyncEpoch() uint64 { + if x != nil { + return x.UserSyncEpoch + } + return 0 +} + // log type Log struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -1386,6 +1410,7 @@ type User struct { Email string `protobuf:"bytes,1,opt,name=email,proto3" json:"email,omitempty"` Proxies *Proxy `protobuf:"bytes,2,opt,name=proxies,proto3" json:"proxies,omitempty"` Inbounds []string `protobuf:"bytes,3,rep,name=inbounds,proto3" json:"inbounds,omitempty"` + UserSyncEpoch uint64 `protobuf:"varint,4,opt,name=user_sync_epoch,json=userSyncEpoch,proto3" json:"user_sync_epoch,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1441,9 +1466,17 @@ func (x *User) GetInbounds() []string { return nil } +func (x *User) GetUserSyncEpoch() uint64 { + if x != nil { + return x.UserSyncEpoch + } + return 0 +} + type Users struct { state protoimpl.MessageState `protogen:"open.v1"` Users []*User `protobuf:"bytes,1,rep,name=users,proto3" json:"users,omitempty"` + UserSyncEpoch uint64 `protobuf:"varint,2,opt,name=user_sync_epoch,json=userSyncEpoch,proto3" json:"user_sync_epoch,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1485,11 +1518,19 @@ func (x *Users) GetUsers() []*User { return nil } +func (x *Users) GetUserSyncEpoch() uint64 { + if x != nil { + return x.UserSyncEpoch + } + return 0 +} + type UsersChunk struct { state protoimpl.MessageState `protogen:"open.v1"` Users []*User `protobuf:"bytes,1,rep,name=users,proto3" json:"users,omitempty"` Index uint64 `protobuf:"varint,2,opt,name=index,proto3" json:"index,omitempty"` Last bool `protobuf:"varint,3,opt,name=last,proto3" json:"last,omitempty"` + UserSyncEpoch uint64 `protobuf:"varint,4,opt,name=user_sync_epoch,json=userSyncEpoch,proto3" json:"user_sync_epoch,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1545,6 +1586,13 @@ func (x *UsersChunk) GetLast() bool { return false } +func (x *UsersChunk) GetUserSyncEpoch() uint64 { + if x != nil { + return x.UserSyncEpoch + } + return 0 +} + // Routing (mirrors xray app/router/command, node-friendly shapes) type RoutingRule struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -2086,18 +2134,21 @@ var File_common_service_proto protoreflect.FileDescriptor const file_common_service_proto_rawDesc = "" + "\n" + "\x14common/service.proto\x12\aservice\"\a\n" + - "\x05Empty\"r\n" + + "\x05Empty\"\xd5\x01\n" + "\x10BaseInfoResponse\x12\x18\n" + "\astarted\x18\x01 \x01(\bR\astarted\x12!\n" + "\fcore_version\x18\x02 \x01(\tR\vcoreVersion\x12!\n" + - "\fnode_version\x18\x03 \x01(\tR\vnodeVersion\"\xba\x01\n" + + "\fnode_version\x18\x03 \x01(\tR\vnodeVersion\x129\n" + + "\x19user_sync_epoch_supported\x18\x04 \x01(\bR\x16userSyncEpochSupported\x12&\n" + + "\x0fuser_sync_epoch\x18\x05 \x01(\x04R\ruserSyncEpoch\"\xe2\x01\n" + "\aBackend\x12(\n" + "\x04type\x18\x01 \x01(\x0e2\x14.service.BackendTypeR\x04type\x12\x16\n" + "\x06config\x18\x02 \x01(\tR\x06config\x12#\n" + "\x05users\x18\x03 \x03(\v2\r.service.UserR\x05users\x12\x1d\n" + "\n" + "keep_alive\x18\x04 \x01(\x04R\tkeepAlive\x12)\n" + - "\x10exclude_inbounds\x18\x05 \x03(\tR\x0fexcludeInbounds\"\x1d\n" + + "\x10exclude_inbounds\x18\x05 \x03(\tR\x0fexcludeInbounds\x12&\n" + + "\x0fuser_sync_epoch\x18\x06 \x01(\x04R\ruserSyncEpoch\"\x1d\n" + "\x03Log\x12\x16\n" + "\x06detail\x18\x01 \x01(\tR\x06detail\"X\n" + "\x04Stat\x12\x12\n" + @@ -2175,18 +2226,21 @@ const file_common_service_proto_rawDesc = "" + "\x06trojan\x18\x03 \x01(\v2\x0f.service.TrojanR\x06trojan\x126\n" + "\vshadowsocks\x18\x04 \x01(\v2\x14.service.ShadowsocksR\vshadowsocks\x120\n" + "\twireguard\x18\x05 \x01(\v2\x12.service.WireguardR\twireguard\x12-\n" + - "\bhysteria\x18\x06 \x01(\v2\x11.service.HysteriaR\bhysteria\"b\n" + + "\bhysteria\x18\x06 \x01(\v2\x11.service.HysteriaR\bhysteria\"\x8a\x01\n" + "\x04User\x12\x14\n" + "\x05email\x18\x01 \x01(\tR\x05email\x12(\n" + "\aproxies\x18\x02 \x01(\v2\x0e.service.ProxyR\aproxies\x12\x1a\n" + - "\binbounds\x18\x03 \x03(\tR\binbounds\",\n" + + "\binbounds\x18\x03 \x03(\tR\binbounds\x12&\n" + + "\x0fuser_sync_epoch\x18\x04 \x01(\x04R\ruserSyncEpoch\"T\n" + "\x05Users\x12#\n" + - "\x05users\x18\x01 \x03(\v2\r.service.UserR\x05users\"[\n" + + "\x05users\x18\x01 \x03(\v2\r.service.UserR\x05users\x12&\n" + + "\x0fuser_sync_epoch\x18\x02 \x01(\x04R\ruserSyncEpoch\"\x83\x01\n" + "\n" + "UsersChunk\x12#\n" + "\x05users\x18\x01 \x03(\v2\r.service.UserR\x05users\x12\x14\n" + "\x05index\x18\x02 \x01(\x04R\x05index\x12\x12\n" + - "\x04last\x18\x03 \x01(\bR\x04last\"K\n" + + "\x04last\x18\x03 \x01(\bR\x04last\x12&\n" + + "\x0fuser_sync_epoch\x18\x04 \x01(\x04R\ruserSyncEpoch\"K\n" + "\vRoutingRule\x12!\n" + "\foutbound_tag\x18\x01 \x01(\tR\voutboundTag\x12\x19\n" + "\brule_tag\x18\x02 \x01(\tR\aruleTag\"B\n" + diff --git a/common/service.proto b/common/service.proto index 486396e..4564d3f 100644 --- a/common/service.proto +++ b/common/service.proto @@ -11,6 +11,8 @@ message BaseInfoResponse { bool started = 1; string core_version = 2; string node_version = 3; + bool user_sync_epoch_supported = 4; + uint64 user_sync_epoch = 5; } enum BackendType { @@ -24,6 +26,7 @@ message Backend { repeated User users = 3; uint64 keep_alive = 4; repeated string exclude_inbounds = 5; + uint64 user_sync_epoch = 6; } // log @@ -150,16 +153,19 @@ message User { string email = 1; Proxy proxies = 2; repeated string inbounds = 3; + uint64 user_sync_epoch = 4; } message Users { repeated User users = 1; + uint64 user_sync_epoch = 2; } message UsersChunk { repeated User users = 1; uint64 index = 2; bool last = 3; + uint64 user_sync_epoch = 4; } // Routing (mirrors xray app/router/command, node-friendly shapes) diff --git a/config/config.go b/config/config.go index 6dd5be8..8f8f58c 100644 --- a/config/config.go +++ b/config/config.go @@ -1,6 +1,7 @@ package config import ( + "fmt" "log" "os" "regexp" @@ -41,37 +42,24 @@ func Load() (*Config, error) { log.Printf("[Warning] Failed to load env file, if you're using 'Docker' and you set 'environment' or 'env_file' variable, don't worry, everything is fine. Error: %v", err) } - cfg := &Config{ - ServicePort: GetEnvAsInt("SERVICE_PORT", 62050), - XrayExecutablePath: GetEnv("XRAY_EXECUTABLE_PATH", "/usr/local/bin/xray"), - XrayAssetsPath: GetEnv("XRAY_ASSETS_PATH", "/usr/local/share/xray"), - SslCertFile: GetEnv("SSL_CERT_FILE", "/var/lib/pg-node/certs/ssl_cert.pem"), - SslKeyFile: GetEnv("SSL_KEY_FILE", "/var/lib/pg-node/certs/ssl_key.pem"), - GeneratedConfigPath: GetEnv("GENERATED_CONFIG_PATH", "/var/lib/pg-node/generated/"), - ServiceProtocol: GetEnv("SERVICE_PROTOCOL", "grpc"), - Debug: GetEnvAsBool("DEBUG", false), - LogBufferSize: GetEnvAsInt("LOG_BUFFER_SIZE", 10000), - StartupLogTailSize: GetEnvAsInt("STARTUP_LOG_TAIL_SIZE", 200), - StatsUpdateIntervalSeconds: GetEnvAsInt("STATS_UPDATE_INTERVAL_SECONDS", 10), - StatsCleanupIntervalSeconds: GetEnvAsInt("STATS_CLEANUP_INTERVAL_SECONDS", 300), - - WGHostRouting: GetEnvAsBool("PG_NODE_WG_HOST_ROUTING", true), - WGNATOutputInterface: GetEnv("PG_NODE_WG_NAT_OUTPUT_INTERFACE", ""), - WGNATEgressOnly: GetEnvAsBool("PG_NODE_WG_NAT_EGRESS_ONLY", true), - WGNATDisable: GetEnvAsBool("PG_NODE_WG_NAT_DISABLE", false), - WGRouteTable: GetEnv("PG_NODE_WG_ROUTE_TABLE", ""), - WGRouteOutInterface: GetEnv("PG_NODE_WG_ROUTE_OUT_INTERFACE", ""), - } + cfg := defaultConfig() if cfg.LogBufferSize <= 0 { log.Printf("[Warning] LOG_BUFFER_SIZE must be greater than 0, got %d. Falling back to 1.", cfg.LogBufferSize) cfg.LogBufferSize = 1 } + if cfg.StatsUpdateIntervalSeconds <= 0 || cfg.StatsCleanupIntervalSeconds <= 0 { + return nil, fmt.Errorf("STATS_UPDATE_INTERVAL_SECONDS and STATS_CLEANUP_INTERVAL_SECONDS must be greater than zero") + } - cfg.ApiKey, err = GetEnvAsUUID("API_KEY") - if err != nil { - log.Printf("[Error] Failed to load API Key, error: %v", err) + apiKey, err := GetEnvAsUUID("API_KEY") + if err != nil || apiKey == uuid.Nil { + if err != nil { + return nil, fmt.Errorf("invalid API_KEY: %w", err) + } + return nil, fmt.Errorf("invalid API_KEY: zero UUID is not allowed") } + cfg.ApiKey = apiKey nodeHostStr := GetEnv("NODE_HOST", "0.0.0.0") ipPattern := `^(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$` @@ -87,9 +75,33 @@ func Load() (*Config, error) { return cfg, nil } +func defaultConfig() *Config { + return &Config{ + ServicePort: GetEnvAsInt("SERVICE_PORT", 62050), + XrayExecutablePath: GetEnv("XRAY_EXECUTABLE_PATH", "/usr/local/bin/xray"), + XrayAssetsPath: GetEnv("XRAY_ASSETS_PATH", "/usr/local/share/xray"), + SslCertFile: GetEnv("SSL_CERT_FILE", "/var/lib/pg-node/certs/ssl_cert.pem"), + SslKeyFile: GetEnv("SSL_KEY_FILE", "/var/lib/pg-node/certs/ssl_key.pem"), + GeneratedConfigPath: GetEnv("GENERATED_CONFIG_PATH", "/var/lib/pg-node/generated/"), + ServiceProtocol: GetEnv("SERVICE_PROTOCOL", "grpc"), + Debug: GetEnvAsBool("DEBUG", false), + LogBufferSize: GetEnvAsInt("LOG_BUFFER_SIZE", 10000), + StartupLogTailSize: GetEnvAsInt("STARTUP_LOG_TAIL_SIZE", 200), + StatsUpdateIntervalSeconds: GetEnvAsInt("STATS_UPDATE_INTERVAL_SECONDS", 10), + StatsCleanupIntervalSeconds: GetEnvAsInt("STATS_CLEANUP_INTERVAL_SECONDS", 300), + + WGHostRouting: GetEnvAsBool("PG_NODE_WG_HOST_ROUTING", true), + WGNATOutputInterface: GetEnv("PG_NODE_WG_NAT_OUTPUT_INTERFACE", ""), + WGNATEgressOnly: GetEnvAsBool("PG_NODE_WG_NAT_EGRESS_ONLY", true), + WGNATDisable: GetEnvAsBool("PG_NODE_WG_NAT_DISABLE", false), + WGRouteTable: GetEnv("PG_NODE_WG_ROUTE_TABLE", ""), + WGRouteOutInterface: GetEnv("PG_NODE_WG_ROUTE_OUT_INTERFACE", ""), + } +} + // NewTestConfig creates a config for testing func NewTestConfig(generatedConfigPath string, key uuid.UUID) *Config { - cfg, _ := Load() + cfg := defaultConfig() cfg.GeneratedConfigPath = generatedConfigPath cfg.ApiKey = key return cfg diff --git a/config/config_test.go b/config/config_test.go new file mode 100644 index 0000000..804f64e --- /dev/null +++ b/config/config_test.go @@ -0,0 +1,37 @@ +package config + +import "testing" + +const validTestAPIKey = "123e4567-e89b-12d3-a456-426614174000" + +func TestLoadRejectsMissingOrUnsafeAPIKey(t *testing.T) { + for _, apiKey := range []string{"", "not-a-uuid", "00000000-0000-0000-0000-000000000000"} { + t.Run(apiKey, func(t *testing.T) { + t.Setenv("API_KEY", apiKey) + if _, err := Load(); err == nil { + t.Fatalf("Load() accepted unsafe API_KEY %q", apiKey) + } + }) + } +} + +func TestLoadRejectsNonPositiveStatsIntervals(t *testing.T) { + t.Setenv("API_KEY", validTestAPIKey) + for _, interval := range []string{"0", "-1"} { + t.Run(interval, func(t *testing.T) { + t.Setenv("STATS_UPDATE_INTERVAL_SECONDS", interval) + if _, err := Load(); err == nil { + t.Fatalf("Load() accepted STATS_UPDATE_INTERVAL_SECONDS=%s", interval) + } + }) + } +} + +func TestLoadAcceptsValidAPIKeyAndIntervals(t *testing.T) { + t.Setenv("API_KEY", validTestAPIKey) + t.Setenv("STATS_UPDATE_INTERVAL_SECONDS", "10") + t.Setenv("STATS_CLEANUP_INTERVAL_SECONDS", "300") + if _, err := Load(); err != nil { + t.Fatalf("Load() returned error: %v", err) + } +} diff --git a/controller/controller.go b/controller/controller.go index 293c892..6dec7d9 100644 --- a/controller/controller.go +++ b/controller/controller.go @@ -3,6 +3,7 @@ package controller import ( "context" "errors" + "fmt" "log" "sync" "time" @@ -25,16 +26,54 @@ type Service interface { } type Controller struct { - backend backend.Backend - cfg *config.Config - apiPort int - metricPort int - clientIP string - lastRequest time.Time - stats *common.SystemStatsResponse - cancelFunc context.CancelFunc - mu sync.RWMutex - controlMu sync.Mutex + backend backend.Backend + cfg *config.Config + apiPort int + metricPort int + clientIP string + lastRequest time.Time + connectionGeneration uint64 + stats *common.SystemStatsResponse + cancelFunc context.CancelFunc + mu sync.RWMutex + controlMu sync.Mutex + userSyncMu sync.Mutex + maxUserSyncEpoch uint64 +} + +// UserSyncEpochError reports a user mutation that is older than one already +// accepted by this node. Epoch zero remains available to legacy clients only +// until the first epoch-aware mutation is accepted. +type UserSyncEpochError struct { + Received uint64 + Current uint64 +} + +// UserSyncEpochBatch validates that every item in one streamed mutation uses +// the same epoch before any backend state is changed. +type UserSyncEpochBatch struct { + epoch uint64 + set bool +} + +func (b *UserSyncEpochBatch) Add(epoch uint64) error { + if !b.set { + b.epoch = epoch + b.set = true + return nil + } + if epoch != b.epoch { + return errors.New("all items must use the same user sync epoch") + } + return nil +} + +func (b *UserSyncEpochBatch) Epoch() uint64 { + return b.epoch +} + +func (e *UserSyncEpochError) Error() string { + return fmt.Sprintf("stale user sync epoch %d; current epoch is %d", e.Received, e.Current) } func New(cfg *config.Config) *Controller { @@ -55,38 +94,48 @@ func (c *Controller) ApiKey() uuid.UUID { func (c *Controller) Connect(ip string, keepAlive uint64) { c.mu.Lock() - defer c.mu.Unlock() c.lastRequest = time.Now() c.clientIP = ip + c.connectionGeneration++ + generation := c.connectionGeneration ctx, cancel := context.WithCancel(context.Background()) c.cancelFunc = cancel + c.mu.Unlock() + go c.recordSystemStats(ctx) if keepAlive > 0 { - go c.keepAliveTracker(ctx, time.Duration(keepAlive)*time.Second) + go c.keepAliveTracker(ctx, time.Duration(keepAlive)*time.Second, generation) } } +// Disconnect serializes an automatic or external disconnect with Start. func (c *Controller) Disconnect() { - c.cancelFunc() + c.LockControl() + defer c.UnlockControl() + c.DisconnectControlled() +} +// DisconnectControlled detaches the current backend before stopping it. The +// caller must hold the controller control lock, which makes replacement and +// keep-alive disconnects mutually exclusive. +func (c *Controller) DisconnectControlled() { c.mu.Lock() + cancel := c.cancelFunc backend := c.backend + c.backend = nil + c.connectionGeneration++ + c.apiPort = netutil.FindFreePort() + c.metricPort = netutil.FindFreePort() + c.clientIP = "" c.mu.Unlock() - // Shutdown backend outside of lock to avoid deadlock - // Shutdown() will wait for process termination to complete + cancel() + // Shutdown may wait for process termination; the detached backend can no + // longer erase a backend that a later Start creates. if backend != nil { backend.Shutdown() } - - c.mu.Lock() - defer c.mu.Unlock() - - c.backend = nil - c.apiPort = netutil.FindFreePort() - c.metricPort = netutil.FindFreePort() - c.clientIP = "" } func (c *Controller) Ip() string { @@ -109,6 +158,29 @@ func (c *Controller) UnlockControl() { c.controlMu.Unlock() } +// ApplyUserSyncEpoch serializes every backend user mutation in epoch order. +// Advancing maxUserSyncEpoch happens before mutation so a failed higher-epoch +// request still fences all older in-flight or retried work. +func (c *Controller) ApplyUserSyncEpoch(epoch uint64, mutate func() error) error { + c.userSyncMu.Lock() + defer c.userSyncMu.Unlock() + + if epoch < c.maxUserSyncEpoch || (epoch == 0 && c.maxUserSyncEpoch > 0) { + return &UserSyncEpochError{Received: epoch, Current: c.maxUserSyncEpoch} + } + if epoch > c.maxUserSyncEpoch { + c.maxUserSyncEpoch = epoch + } + + return mutate() +} + +func (c *Controller) UserSyncEpoch() uint64 { + c.userSyncMu.Lock() + defer c.userSyncMu.Unlock() + return c.maxUserSyncEpoch +} + func (c *Controller) NewRequest() { c.mu.Lock() defer c.mu.Unlock() @@ -156,13 +228,30 @@ func (c *Controller) StartBackend(ctx context.Context, backend *common.Backend) return nil } +// StartBackendControlled replaces the current backend only after reserving the +// request epoch. The caller must hold the control lock and verify ownership. +func (c *Controller) StartBackendControlled(ctx context.Context, data *common.Backend, clientIP string) error { + return c.ApplyUserSyncEpoch(data.GetUserSyncEpoch(), func() error { + if c.Backend() != nil { + log.Println("New connection from ", clientIP, " core control access was taken away from previous client.") + c.DisconnectControlled() + } + + if err := c.StartBackend(ctx, data); err != nil { + return err + } + c.Connect(clientIP, data.GetKeepAlive()) + return nil + }) +} + func (c *Controller) Backend() backend.Backend { c.mu.RLock() defer c.mu.RUnlock() return c.backend } -func (c *Controller) keepAliveTracker(ctx context.Context, keepAlive time.Duration) { +func (c *Controller) keepAliveTracker(ctx context.Context, keepAlive time.Duration, generation uint64) { ticker := time.NewTicker(5 * time.Second) defer ticker.Stop() @@ -171,17 +260,28 @@ func (c *Controller) keepAliveTracker(ctx context.Context, keepAlive time.Durati case <-ctx.Done(): return case <-ticker.C: - c.mu.RLock() - lastRequest := c.lastRequest - c.mu.RUnlock() - if time.Since(lastRequest) >= keepAlive { + if c.keepAliveExpired(generation, keepAlive) { log.Println("disconnect automatically due to keep alive timeout") - c.Disconnect() + c.LockControl() + if c.keepAliveExpired(generation, keepAlive) { + c.DisconnectControlled() + } + c.UnlockControl() } } } } +// keepAliveExpired verifies that the tracker still owns the current connection. +// A previous tracker may already have observed its timeout while Start is replacing +// the backend, so it must re-check its generation under the control lock before +// disconnecting anything. +func (c *Controller) keepAliveExpired(generation uint64, keepAlive time.Duration) bool { + c.mu.RLock() + defer c.mu.RUnlock() + return c.connectionGeneration == generation && time.Since(c.lastRequest) >= keepAlive +} + func (c *Controller) recordSystemStats(ctx context.Context) { interval := 1500 * time.Millisecond @@ -247,13 +347,17 @@ func (c *Controller) SystemStats(ctx context.Context) *common.SystemStatsRespons } func (c *Controller) BaseInfoResponse() *common.BaseInfoResponse { + userSyncEpoch := c.UserSyncEpoch() + c.mu.Lock() defer c.mu.Unlock() response := &common.BaseInfoResponse{ - Started: false, - CoreVersion: "", - NodeVersion: NodeVersion, + Started: false, + CoreVersion: "", + NodeVersion: NodeVersion, + UserSyncEpochSupported: true, + UserSyncEpoch: userSyncEpoch, } if c.backend != nil { diff --git a/controller/controller_lifecycle_test.go b/controller/controller_lifecycle_test.go new file mode 100644 index 0000000..d7511a7 --- /dev/null +++ b/controller/controller_lifecycle_test.go @@ -0,0 +1,92 @@ +package controller + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/pasarguard/node/backend" + "github.com/pasarguard/node/common" + "github.com/pasarguard/node/config" +) + +type blockingBackend struct { + shutdownStarted chan struct{} + allowShutdown chan struct{} +} + +func (b *blockingBackend) Started() bool { return true } +func (b *blockingBackend) Version() string { return "test" } +func (b *blockingBackend) Logs() <-chan string { return nil } +func (b *blockingBackend) Restart() error { return nil } +func (b *blockingBackend) Shutdown() { + close(b.shutdownStarted) + <-b.allowShutdown +} +func (b *blockingBackend) SyncUser(context.Context, *common.User) error { return nil } +func (b *blockingBackend) SyncUsers(context.Context, []*common.User) error { return nil } +func (b *blockingBackend) UpdateUsers(context.Context, []*common.User) error { return nil } +func (b *blockingBackend) UpdateUsersAndRestart(context.Context, []*common.User) error { return nil } +func (b *blockingBackend) GetSysStats(context.Context) (*common.BackendStatsResponse, error) { + return nil, nil +} +func (b *blockingBackend) GetStats(context.Context, *common.StatRequest) (*common.StatResponse, error) { + return nil, nil +} +func (b *blockingBackend) GetOutboundsLatency(context.Context, *common.LatencyRequest) (*common.LatencyResponse, error) { + return nil, nil +} +func (b *blockingBackend) GetUserOnlineStats(context.Context, string) (*common.OnlineStatResponse, error) { + return nil, nil +} +func (b *blockingBackend) GetUserOnlineIpListStats(context.Context, string) (*common.StatsOnlineIpListResponse, error) { + return nil, nil +} + +var _ backend.Backend = (*blockingBackend)(nil) + +func TestDisconnectSerializesBackendReplacement(t *testing.T) { + c := New(config.NewTestConfig(t.TempDir(), uuid.New())) + oldBackend := &blockingBackend{shutdownStarted: make(chan struct{}), allowShutdown: make(chan struct{})} + newBackend := &blockingBackend{shutdownStarted: make(chan struct{}), allowShutdown: make(chan struct{})} + c.backend = oldBackend + + disconnected := make(chan struct{}) + go func() { + c.Disconnect() + close(disconnected) + }() + + select { + case <-oldBackend.shutdownStarted: + case <-time.After(time.Second): + t.Fatal("Disconnect did not begin shutting down the old backend") + } + + replacementStarted := make(chan struct{}) + replacementDone := make(chan struct{}) + go func() { + c.LockControl() + close(replacementStarted) + c.backend = newBackend + c.UnlockControl() + close(replacementDone) + }() + + select { + case <-replacementStarted: + t.Fatal("replacement acquired control while Disconnect was still stopping the old backend") + case <-time.After(50 * time.Millisecond): + } + + close(oldBackend.allowShutdown) + <-disconnected + <-replacementStarted + <-replacementDone + + if c.Backend() != newBackend { + t.Fatal("stale Disconnect erased the replacement backend") + } +} diff --git a/controller/controller_test.go b/controller/controller_test.go new file mode 100644 index 0000000..77ed04a --- /dev/null +++ b/controller/controller_test.go @@ -0,0 +1,31 @@ +package controller + +import ( + "testing" + "time" + + "github.com/pasarguard/node/config" +) + +func TestStaleKeepAliveCannotDisconnectReplacementConnection(t *testing.T) { + c := New(&config.Config{}) + c.Connect("old-client", 0) + + c.mu.RLock() + staleGeneration := c.connectionGeneration + c.mu.RUnlock() + + c.Connect("new-client", 0) + if c.keepAliveExpired(staleGeneration, 0) { + t.Fatal("a stale keep-alive tracker must not consider a replacement connection expired") + } + if got := c.Ip(); got != "new-client" { + t.Fatalf("replacement connection was changed: got %q", got) + } + + // The stale tracker has already decided its old lease timed out. It must + // still be rejected after Start replaces the connection. + if c.keepAliveExpired(staleGeneration, time.Nanosecond) { + t.Fatal("stale keep-alive generation became valid after replacement") + } +} diff --git a/controller/rest/base.go b/controller/rest/base.go index 3858d26..26fd864 100644 --- a/controller/rest/base.go +++ b/controller/rest/base.go @@ -1,7 +1,7 @@ package rest import ( - "log" + "errors" "net/http" "github.com/pasarguard/node/common" @@ -18,6 +18,10 @@ func (s *Service) Start(w http.ResponseWriter, r *http.Request) { data := &common.Backend{} if err := common.ReadProtoBody(r.Body, data); err != nil { + if errors.Is(err, common.ErrProtoBodyTooLarge) { + http.Error(w, err.Error(), http.StatusRequestEntityTooLarge) + return + } http.Error(w, err.Error(), http.StatusBadRequest) return } @@ -28,22 +32,16 @@ func (s *Service) Start(w http.ResponseWriter, r *http.Request) { return } - if s.Backend() != nil { - if !s.IsCurrentClient(ip) { - http.Error(w, "node is controlled by another client", http.StatusForbidden) - return - } - log.Println("New connection from ", ip, " core control access was taken away from previous client.") - s.Disconnect() + if s.Backend() != nil && !s.IsCurrentClient(ip) { + http.Error(w, "node is controlled by another client", http.StatusForbidden) + return } - if err := s.StartBackend(r.Context(), data); err != nil { - http.Error(w, err.Error(), http.StatusServiceUnavailable) + if err := s.StartBackendControlled(r.Context(), data, ip); err != nil { + writeUserSyncError(w, err, http.StatusServiceUnavailable) return } - s.Connect(ip, data.GetKeepAlive()) - common.SendProtoResponse(w, s.BaseInfoResponse()) } @@ -51,7 +49,7 @@ func (s *Service) Stop(w http.ResponseWriter, _ *http.Request) { s.LockControl() defer s.UnlockControl() - s.Disconnect() + s.DisconnectControlled() common.SendProtoResponse(w, &common.Empty{}) } diff --git a/controller/rest/http_timeout.go b/controller/rest/http_timeout.go new file mode 100644 index 0000000..d5c54f0 --- /dev/null +++ b/controller/rest/http_timeout.go @@ -0,0 +1,76 @@ +package rest + +import ( + "context" + "crypto/tls" + "fmt" + "net/http" + "time" +) + +const responseWriteTimeout = 30 * time.Second + +func newHTTPServer(tlsConfig *tls.Config, addr string, handler http.Handler) *http.Server { + return &http.Server{ + Addr: addr, + TLSConfig: tlsConfig, + Handler: handler, + ReadHeaderTimeout: 10 * time.Second, + ReadTimeout: 30 * time.Second, + WriteTimeout: responseWriteTimeout, + IdleTimeout: 120 * time.Second, + MaxHeaderBytes: 16 * 1024, + } +} + +func disableWriteDeadline(w http.ResponseWriter) error { + return http.NewResponseController(w).SetWriteDeadline(time.Time{}) +} + +func stopWritesOnContext(ctx context.Context, w http.ResponseWriter) func() { + controller := http.NewResponseController(w) + cancelDeadlineDone := make(chan struct{}) + stopCancelDeadline := context.AfterFunc(ctx, func() { + _ = controller.SetWriteDeadline(time.Now()) + close(cancelDeadlineDone) + }) + return func() { + if !stopCancelDeadline() { + <-cancelDeadlineDone + } + } +} + +func writeLogLine(ctx context.Context, w http.ResponseWriter, line string, timeout time.Duration) error { + if err := ctx.Err(); err != nil { + return err + } + + controller := http.NewResponseController(w) + if err := controller.SetWriteDeadline(time.Now().Add(timeout)); err != nil { + return err + } + // Cancellation may race with the sliding deadline update above. Re-check + // after setting it so an already-expired cancellation deadline can never be + // overwritten by the longer per-write timeout. + if err := ctx.Err(); err != nil { + _ = controller.SetWriteDeadline(time.Now()) + return err + } + if _, err := fmt.Fprintf(w, "%s\n", line); err != nil { + if ctxErr := ctx.Err(); ctxErr != nil { + return ctxErr + } + return err + } + if err := controller.Flush(); err != nil { + if ctxErr := ctx.Err(); ctxErr != nil { + return ctxErr + } + return err + } + if err := controller.SetWriteDeadline(time.Time{}); err != nil { + return err + } + return ctx.Err() +} diff --git a/controller/rest/log.go b/controller/rest/log.go index 3254b78..7286c5f 100644 --- a/controller/rest/log.go +++ b/controller/rest/log.go @@ -1,16 +1,19 @@ package rest -import ( - "fmt" - "net/http" -) +import "net/http" func (s *Service) GetLogs(w http.ResponseWriter, r *http.Request) { - flusher, ok := w.(http.Flusher) + _, ok := w.(http.Flusher) if !ok { http.Error(w, "Streaming unsupported", http.StatusInternalServerError) return } + if err := disableWriteDeadline(w); err != nil { + http.Error(w, "Streaming deadline control unsupported", http.StatusInternalServerError) + return + } + stopContextWrites := stopWritesOnContext(r.Context(), w) + defer stopContextWrites() w.Header().Set("Content-Type", "text/event-stream") w.Header().Set("Cache-Control", "no-cache") @@ -25,13 +28,10 @@ func (s *Service) GetLogs(w http.ResponseWriter, r *http.Request) { return } - _, err := fmt.Fprintf(w, "%s\n", log) - if err != nil { + if err := writeLogLine(r.Context(), w, log, responseWriteTimeout); err != nil { return } - flusher.Flush() - case <-r.Context().Done(): return } diff --git a/controller/rest/log_timeout_test.go b/controller/rest/log_timeout_test.go new file mode 100644 index 0000000..1397b29 --- /dev/null +++ b/controller/rest/log_timeout_test.go @@ -0,0 +1,185 @@ +package rest + +import ( + "context" + "errors" + "io" + "net" + "net/http" + "testing" + "time" + + "github.com/go-chi/chi/v5/middleware" +) + +type pipeResponseWriter struct { + conn net.Conn + header http.Header +} + +type cancelOnDeadlineWriter struct { + header http.Header + cancel context.CancelFunc + deadlineSet int + wrote bool +} + +func (w *cancelOnDeadlineWriter) Header() http.Header { return w.header } +func (w *cancelOnDeadlineWriter) WriteHeader(int) {} +func (w *cancelOnDeadlineWriter) Write(payload []byte) (int, error) { + w.wrote = true + return len(payload), nil +} +func (w *cancelOnDeadlineWriter) SetWriteDeadline(time.Time) error { + w.deadlineSet++ + if w.deadlineSet == 1 { + w.cancel() + } + return nil +} + +func newPipeResponseWriter(conn net.Conn) *pipeResponseWriter { + return &pipeResponseWriter{conn: conn, header: make(http.Header)} +} + +func (w *pipeResponseWriter) Header() http.Header { + return w.header +} + +func (w *pipeResponseWriter) Write(payload []byte) (int, error) { + return w.conn.Write(payload) +} + +func (w *pipeResponseWriter) WriteHeader(int) {} + +func (w *pipeResponseWriter) Flush() {} + +func (w *pipeResponseWriter) SetWriteDeadline(deadline time.Time) error { + return w.conn.SetWriteDeadline(deadline) +} + +func wrappedPipeResponseWriter(conn net.Conn) http.ResponseWriter { + w := middleware.NewWrapResponseWriter(newPipeResponseWriter(conn), 1) + return middleware.NewWrapResponseWriter(w, 1) +} + +func writeAndReadLogLine(t *testing.T, w http.ResponseWriter, reader net.Conn, line string, timeout time.Duration) { + t.Helper() + + want := []byte(line + "\n") + readResult := make(chan error, 1) + go func() { + if err := reader.SetReadDeadline(time.Now().Add(time.Second)); err != nil { + readResult <- err + return + } + got := make([]byte, len(want)) + if _, err := io.ReadFull(reader, got); err != nil { + readResult <- err + return + } + if string(got) != string(want) { + readResult <- errors.New("unexpected log line") + return + } + readResult <- nil + }() + + if err := writeLogLine(context.Background(), w, line, timeout); err != nil { + t.Fatalf("write log line: %v", err) + } + if err := <-readResult; err != nil { + t.Fatalf("read log line: %v", err) + } +} + +func TestLogStreamSurvivesIdlePeriodsPastWriteTimeout(t *testing.T) { + serverConn, clientConn := net.Pipe() + defer serverConn.Close() + defer clientConn.Close() + + w := wrappedPipeResponseWriter(serverConn) + timeout := 50 * time.Millisecond + if err := serverConn.SetWriteDeadline(time.Now().Add(timeout)); err != nil { + t.Fatalf("set initial server deadline: %v", err) + } + if err := disableWriteDeadline(w); err != nil { + t.Fatalf("disable idle write deadline: %v", err) + } + + time.Sleep(3 * timeout) + writeAndReadLogLine(t, w, clientConn, "after initial idle", timeout) + + // A successful write must clear its sliding deadline while the handler waits + // for the next log entry. + time.Sleep(3 * timeout) + writeAndReadLogLine(t, w, clientConn, "after second idle", timeout) +} + +func TestLogStreamBoundsStalledReader(t *testing.T) { + serverConn, clientConn := net.Pipe() + defer serverConn.Close() + defer clientConn.Close() + + timeout := 50 * time.Millisecond + started := time.Now() + err := writeLogLine(context.Background(), wrappedPipeResponseWriter(serverConn), "blocked", timeout) + elapsed := time.Since(started) + + if err == nil { + t.Fatal("write unexpectedly succeeded with a stalled reader") + } + var netErr net.Error + if !errors.As(err, &netErr) || !netErr.Timeout() { + t.Fatalf("write error = %v, want network timeout", err) + } + if elapsed > 10*timeout { + t.Fatalf("stalled write took %v, want at most %v", elapsed, 10*timeout) + } +} + +func TestLogStreamCancelsStalledWriteWithRequestContext(t *testing.T) { + serverConn, clientConn := net.Pipe() + defer serverConn.Close() + defer clientConn.Close() + + ctx, cancel := context.WithCancel(context.Background()) + w := wrappedPipeResponseWriter(serverConn) + stopContextWrites := stopWritesOnContext(ctx, w) + defer stopContextWrites() + time.AfterFunc(50*time.Millisecond, cancel) + started := time.Now() + err := writeLogLine(ctx, w, "blocked", time.Second) + elapsed := time.Since(started) + + if !errors.Is(err, context.Canceled) { + t.Fatalf("write error = %v, want context.Canceled", err) + } + if elapsed > 500*time.Millisecond { + t.Fatalf("canceled write took %v, want at most 500ms", elapsed) + } +} + +func TestLogStreamCancellationCannotBeOverwrittenBySlidingDeadline(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + w := &cancelOnDeadlineWriter{header: make(http.Header), cancel: cancel} + + err := writeLogLine(ctx, w, "must not be written", time.Second) + + if !errors.Is(err, context.Canceled) { + t.Fatalf("write error = %v, want context.Canceled", err) + } + if w.wrote { + t.Fatal("write started after cancellation raced with the sliding deadline") + } + if w.deadlineSet != 2 { + t.Fatalf("deadline updates = %d, want sliding deadline followed by immediate cancellation", w.deadlineSet) + } +} + +func TestHTTPServerKeepsWriteTimeoutForRegularResponses(t *testing.T) { + server := newHTTPServer(nil, "127.0.0.1:0", http.NotFoundHandler()) + if server.WriteTimeout != responseWriteTimeout { + t.Fatalf("WriteTimeout = %v, want %v", server.WriteTimeout, responseWriteTimeout) + } +} diff --git a/controller/rest/middleware.go b/controller/rest/middleware.go index c1f2a69..1a9f4bd 100644 --- a/controller/rest/middleware.go +++ b/controller/rest/middleware.go @@ -19,6 +19,10 @@ func (s *Service) validateApiKey(next http.Handler) http.Handler { // check API key apiKey := s.ApiKey() + if apiKey == uuid.Nil { + http.Error(w, "node API key is not configured", http.StatusServiceUnavailable) + return + } key, err := uuid.Parse(apiKeyHeader) switch { diff --git a/controller/rest/service.go b/controller/rest/service.go index fc9aa6d..8998b60 100644 --- a/controller/rest/service.go +++ b/controller/rest/service.go @@ -32,12 +32,12 @@ func (s *Service) setRouter() { router.Use(middleware.Recoverer) router.Post("/start", s.Start) + router.Put("/stop", s.Stop) router.Get("/info", s.Base) router.Group(func(private chi.Router) { private.Use(s.checkBackendMiddleware) - private.Put("/stop", s.Stop) private.Get("/logs", s.GetLogs) // stats api private.Route("/stats", func(statsGroup chi.Router) { @@ -77,11 +77,7 @@ type Service struct { func StartHttpListener(tlsConfig *tls.Config, addr string, cfg *config.Config) (func(ctx context.Context) error, controller.Service, error) { s := New(cfg) - httpServer := &http.Server{ - Addr: addr, - TLSConfig: tlsConfig, - Handler: s.Router, - } + httpServer := newHTTPServer(tlsConfig, addr, s.Router) // Test if we can listen on the port before starting the goroutine listener, err := tls.Listen("tcp", addr, tlsConfig) diff --git a/controller/rest/user.go b/controller/rest/user.go index 4f6bf3d..f06a226 100644 --- a/controller/rest/user.go +++ b/controller/rest/user.go @@ -15,16 +15,32 @@ import ( "github.com/pasarguard/node/controller" ) -func (s *Service) SyncUser(w http.ResponseWriter, r *http.Request) { +const maxChunkBytes uint64 = 8 * 1024 * 1024 + +func readRequestBody(w http.ResponseWriter, r *http.Request) ([]byte, bool) { + r.Body = http.MaxBytesReader(w, r.Body, common.MaxProtoBodyBytes) + defer r.Body.Close() body, err := io.ReadAll(r.Body) - if err != nil { + if err == nil { + return body, true + } + var maxBytesError *http.MaxBytesError + if errors.As(err, &maxBytesError) { + http.Error(w, "request body too large", http.StatusRequestEntityTooLarge) + } else { http.Error(w, "Failed to read request body", http.StatusBadRequest) + } + return nil, false +} + +func (s *Service) SyncUser(w http.ResponseWriter, r *http.Request) { + body, ok := readRequestBody(w, r) + if !ok { return } - defer r.Body.Close() user := &common.User{} - if err = proto.Unmarshal(body, user); err != nil { + if err := proto.Unmarshal(body, user); err != nil { http.Error(w, "Failed to decode user", http.StatusBadRequest) return } @@ -36,57 +52,69 @@ func (s *Service) SyncUser(w http.ResponseWriter, r *http.Request) { log.Printf("Got user: %v", user.GetEmail()) - if err = s.Backend().SyncUser(r.Context(), user); err != nil { + if err := s.ApplyUserSyncEpoch(user.GetUserSyncEpoch(), func() error { + back := s.Backend() + if back == nil { + return errors.New("backend is not started") + } + return back.SyncUser(r.Context(), user) + }); err != nil { log.Printf("Error syncing user: %v", err) - http.Error(w, err.Error(), http.StatusInternalServerError) + writeUserSyncError(w, err, http.StatusInternalServerError) return } response, _ := proto.Marshal(&common.Empty{}) w.Header().Set("Content-Type", "application/x-protobuf") - if _, err = w.Write(response); err != nil { + if _, err := w.Write(response); err != nil { http.Error(w, "Failed to write response", http.StatusInternalServerError) return } } func (s *Service) SyncUsers(w http.ResponseWriter, r *http.Request) { - body, err := io.ReadAll(r.Body) - if err != nil { - http.Error(w, "Failed to read request body", http.StatusBadRequest) + body, ok := readRequestBody(w, r) + if !ok { return } - defer r.Body.Close() users := &common.Users{} - if err = proto.Unmarshal(body, users); err != nil { + if err := proto.Unmarshal(body, users); err != nil { http.Error(w, "Failed to decode user", http.StatusBadRequest) return } - if err = s.Backend().SyncUsers(r.Context(), users.GetUsers()); err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) + if err := s.ApplyUserSyncEpoch(users.GetUserSyncEpoch(), func() error { + back := s.Backend() + if back == nil { + return errors.New("backend is not started") + } + return back.SyncUsers(r.Context(), users.GetUsers()) + }); err != nil { + writeUserSyncError(w, err, http.StatusInternalServerError) return } response, _ := proto.Marshal(&common.Empty{}) w.Header().Set("Content-Type", "application/x-protobuf") - if _, err = w.Write(response); err != nil { + if _, err := w.Write(response); err != nil { http.Error(w, "Failed to write response", http.StatusInternalServerError) return } } func (s *Service) SyncUsersChunked(w http.ResponseWriter, r *http.Request) { + r.Body = http.MaxBytesReader(w, r.Body, common.MaxProtoBodyBytes) reader := bufio.NewReader(r.Body) defer r.Body.Close() chunks := make(map[uint64][]*common.User) var ( - lastIndex uint64 - sawLast bool + lastIndex uint64 + sawLast bool + epochBatch controller.UserSyncEpochBatch ) for { @@ -101,8 +129,12 @@ func (s *Service) SyncUsersChunked(w http.ResponseWriter, r *http.Request) { if size == 0 { continue } + if size > maxChunkBytes || size > uint64(common.MaxProtoBodyBytes) { + http.Error(w, "chunk payload too large", http.StatusRequestEntityTooLarge) + return + } - payload := make([]byte, size) + payload := make([]byte, int(size)) if _, err = io.ReadFull(reader, payload); err != nil { http.Error(w, fmt.Sprintf("failed to read chunk payload: %v", err), http.StatusBadRequest) return @@ -113,6 +145,10 @@ func (s *Service) SyncUsersChunked(w http.ResponseWriter, r *http.Request) { http.Error(w, fmt.Sprintf("failed to decode chunk: %v", err), http.StatusBadRequest) return } + if err = epochBatch.Add(chunk.GetUserSyncEpoch()); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } chunks[chunk.GetIndex()] = append(chunks[chunk.GetIndex()], chunk.GetUsers()...) @@ -129,8 +165,14 @@ func (s *Service) SyncUsersChunked(w http.ResponseWriter, r *http.Request) { return } - if err := controller.ApplyChunkedUserUpdate(r.Context(), s.Backend(), users); err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) + if err := s.ApplyUserSyncEpoch(epochBatch.Epoch(), func() error { + back := s.Backend() + if back == nil { + return errors.New("backend is not started") + } + return controller.ApplyChunkedUserUpdate(r.Context(), back, users) + }); err != nil { + writeUserSyncError(w, err, http.StatusInternalServerError) return } diff --git a/controller/rest/user_sync_epoch.go b/controller/rest/user_sync_epoch.go new file mode 100644 index 0000000..7f6709a --- /dev/null +++ b/controller/rest/user_sync_epoch.go @@ -0,0 +1,17 @@ +package rest + +import ( + "errors" + "net/http" + + "github.com/pasarguard/node/controller" +) + +func writeUserSyncError(w http.ResponseWriter, err error, fallbackStatus int) { + var epochErr *controller.UserSyncEpochError + if errors.As(err, &epochErr) { + http.Error(w, epochErr.Error(), http.StatusPreconditionFailed) + return + } + http.Error(w, err.Error(), fallbackStatus) +} diff --git a/controller/rest/user_sync_epoch_test.go b/controller/rest/user_sync_epoch_test.go new file mode 100644 index 0000000..e11628d --- /dev/null +++ b/controller/rest/user_sync_epoch_test.go @@ -0,0 +1,66 @@ +package rest + +import ( + "bytes" + "encoding/binary" + "net/http" + "net/http/httptest" + "testing" + + "github.com/google/uuid" + "github.com/pasarguard/node/common" + "github.com/pasarguard/node/config" + "github.com/pasarguard/node/controller" + "google.golang.org/protobuf/proto" +) + +func TestRESTChunkedSyncRejectsMixedEpochs(t *testing.T) { + service := &Service{Controller: *controller.New(config.NewTestConfig(t.TempDir(), uuid.New()))} + var body bytes.Buffer + for _, chunk := range []*common.UsersChunk{ + {Index: 0, UserSyncEpoch: 40, Users: []*common.User{{Email: "first@example.com"}}}, + {Index: 1, Last: true, UserSyncEpoch: 41, Users: []*common.User{{Email: "second@example.com"}}}, + } { + payload, err := proto.Marshal(chunk) + if err != nil { + t.Fatal(err) + } + var length [binary.MaxVarintLen64]byte + n := binary.PutUvarint(length[:], uint64(len(payload))) + body.Write(length[:n]) + body.Write(payload) + } + + req := httptest.NewRequest(http.MethodPut, "/users/sync/chunked", &body) + recorder := httptest.NewRecorder() + service.SyncUsersChunked(recorder, req) + + if recorder.Code != http.StatusBadRequest { + t.Fatalf("expected 400 for mixed epochs, got %d: %s", recorder.Code, recorder.Body.String()) + } +} + +func TestRESTUserSyncEpochErrorUsesPreconditionFailed(t *testing.T) { + recorder := httptest.NewRecorder() + writeUserSyncError(recorder, &controller.UserSyncEpochError{Received: 2, Current: 3}, http.StatusInternalServerError) + if recorder.Code != http.StatusPreconditionFailed { + t.Fatalf("expected 412, got %d", recorder.Code) + } +} + +func TestRESTStaleUserSyncRejectsBeforeBackendAccess(t *testing.T) { + service := &Service{Controller: *controller.New(config.NewTestConfig(t.TempDir(), uuid.New()))} + if err := service.ApplyUserSyncEpoch(70, func() error { return nil }); err != nil { + t.Fatalf("failed to establish current epoch: %v", err) + } + payload, err := proto.Marshal(&common.User{Email: "stale@example.com", UserSyncEpoch: 69}) + if err != nil { + t.Fatal(err) + } + req := httptest.NewRequest(http.MethodPut, "/user/sync", bytes.NewReader(payload)) + recorder := httptest.NewRecorder() + service.SyncUser(recorder, req) + if recorder.Code != http.StatusPreconditionFailed { + t.Fatalf("expected stale request to fail with 412 before nil backend access, got %d: %s", recorder.Code, recorder.Body.String()) + } +} diff --git a/controller/rpc/base.go b/controller/rpc/base.go index ec2ca53..1fb2ef6 100644 --- a/controller/rpc/base.go +++ b/controller/rpc/base.go @@ -2,7 +2,6 @@ package rpc import ( "context" - "log" "github.com/pasarguard/node/common" "google.golang.org/grpc/codes" @@ -18,20 +17,14 @@ func (s *Service) Start(ctx context.Context, data *common.Backend) (*common.Base return nil, status.Errorf(codes.PermissionDenied, "unknown client ip") } - if s.Backend() != nil { - if !s.IsCurrentClient(clientIP) { - return nil, status.Errorf(codes.PermissionDenied, "node is controlled by another client") - } - log.Println("New connection from ", clientIP, " core control access was taken away from previous client.") - s.Disconnect() + if s.Backend() != nil && !s.IsCurrentClient(clientIP) { + return nil, status.Errorf(codes.PermissionDenied, "node is controlled by another client") } - if err := s.StartBackend(ctx, data); err != nil { - return nil, err + if err := s.StartBackendControlled(ctx, data, clientIP); err != nil { + return nil, userSyncError(err) } - s.Connect(clientIP, data.GetKeepAlive()) - return s.BaseInfoResponse(), nil } @@ -39,8 +32,8 @@ func (s *Service) Stop(_ context.Context, _ *common.Empty) (*common.Empty, error s.LockControl() defer s.UnlockControl() - s.Disconnect() - return nil, nil + s.DisconnectControlled() + return &common.Empty{}, nil } func (s *Service) GetBaseInfo(_ context.Context, _ *common.Empty) (*common.BaseInfoResponse, error) { diff --git a/controller/rpc/middleware.go b/controller/rpc/middleware.go index 95fe495..e561725 100644 --- a/controller/rpc/middleware.go +++ b/controller/rpc/middleware.go @@ -47,6 +47,9 @@ func validateApiKey(ctx context.Context, s *Service) error { apiKeyHeader := apiKeys[0] apiKey := s.ApiKey() + if apiKey == uuid.Nil { + return status.Errorf(codes.Unavailable, "node API key is not configured") + } key, err := uuid.Parse(apiKeyHeader) switch { case err != nil: @@ -205,7 +208,6 @@ var backendMethods = map[string]bool{ "/service.NodeService/GetUserOnlineIpListStats": true, "/service.NodeService/GetBackendStats": true, "/service.NodeService/GetSystemStats": true, - "/service.NodeService/Stop": true, "/service.NodeService/SyncUser": true, "/service.NodeService/SyncUsers": true, "/service.NodeService/SyncUsersChunked": true, diff --git a/controller/rpc/user.go b/controller/rpc/user.go index 1a275c2..4bf1f1c 100644 --- a/controller/rpc/user.go +++ b/controller/rpc/user.go @@ -9,54 +9,93 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" "github.com/pasarguard/node/common" "github.com/pasarguard/node/controller" ) -func (s *Service) SyncUser(stream grpc.ClientStreamingServer[common.User, common.Empty]) error { - backend, err := s.backend() - if err != nil { - return err +func addUserSyncStreamPayload(total int64, message proto.Message) (int64, error) { + next := total + int64(proto.Size(message)) + if next > common.MaxProtoBodyBytes { + return total, status.Error(codes.ResourceExhausted, "user sync stream payload too large") } + return next, nil +} + +func (s *Service) SyncUser(stream grpc.ClientStreamingServer[common.User, common.Empty]) error { + users := make([]*common.User, 0) + var epochBatch controller.UserSyncEpochBatch + var streamBytes int64 for { user, err := stream.Recv() + if errors.Is(err, io.EOF) { + break + } if err != nil { - return stream.SendAndClose(&common.Empty{}) + return status.Errorf(codes.Internal, "failed to receive user: %v", err) } if user.GetEmail() == "" { return errors.New("email is required") } + streamBytes, err = addUserSyncStreamPayload(streamBytes, user) + if err != nil { + return err + } + if err = epochBatch.Add(user.GetUserSyncEpoch()); err != nil { + return status.Error(codes.InvalidArgument, err.Error()) + } log.Printf("Got user: %v", user.GetEmail()) + users = append(users, user) + } - if err = backend.SyncUser(stream.Context(), user); err != nil { - log.Printf("Error syncing user: %v", err) - return status.Errorf(codes.Internal, "failed to update user: %v", err) + if len(users) == 0 { + return stream.SendAndClose(&common.Empty{}) + } + + if err := s.ApplyUserSyncEpoch(epochBatch.Epoch(), func() error { + back, err := s.backend() + if err != nil { + return err + } + for _, user := range users { + if err = back.SyncUser(stream.Context(), user); err != nil { + log.Printf("Error syncing user: %v", err) + return status.Errorf(codes.Internal, "failed to update user: %v", err) + } } + return nil + }); err != nil { + return userSyncError(err) } + + return stream.SendAndClose(&common.Empty{}) } func (s *Service) SyncUsers(ctx context.Context, users *common.Users) (*common.Empty, error) { - backend, err := s.backend() - if err != nil { - return nil, err - } - - if err := backend.SyncUsers(ctx, users.GetUsers()); err != nil { - return nil, err + if err := s.ApplyUserSyncEpoch(users.GetUserSyncEpoch(), func() error { + back, err := s.backend() + if err != nil { + return err + } + return back.SyncUsers(ctx, users.GetUsers()) + }); err != nil { + return nil, userSyncError(err) } - return nil, nil + return &common.Empty{}, nil } func (s *Service) SyncUsersChunked(stream grpc.ClientStreamingServer[common.UsersChunk, common.Empty]) error { chunks := make(map[uint64][]*common.User) var ( - lastIndex uint64 - sawLast bool + lastIndex uint64 + sawLast bool + epochBatch controller.UserSyncEpochBatch + streamBytes int64 ) for { @@ -67,6 +106,13 @@ func (s *Service) SyncUsersChunked(stream grpc.ClientStreamingServer[common.User if err != nil { return status.Errorf(codes.Internal, "failed to receive chunk: %v", err) } + streamBytes, err = addUserSyncStreamPayload(streamBytes, chunk) + if err != nil { + return err + } + if err = epochBatch.Add(chunk.GetUserSyncEpoch()); err != nil { + return status.Error(codes.InvalidArgument, err.Error()) + } chunks[chunk.GetIndex()] = append(chunks[chunk.GetIndex()], chunk.GetUsers()...) @@ -82,13 +128,17 @@ func (s *Service) SyncUsersChunked(stream grpc.ClientStreamingServer[common.User return status.Error(codes.InvalidArgument, err.Error()) } - backend, err := s.backend() - if err != nil { - return err - } - - if err := controller.ApplyChunkedUserUpdate(stream.Context(), backend, users); err != nil { - return status.Errorf(codes.Internal, "failed to update users: %v", err) + if err := s.ApplyUserSyncEpoch(epochBatch.Epoch(), func() error { + back, err := s.backend() + if err != nil { + return err + } + if err = controller.ApplyChunkedUserUpdate(stream.Context(), back, users); err != nil { + return status.Errorf(codes.Internal, "failed to update users: %v", err) + } + return nil + }); err != nil { + return userSyncError(err) } return stream.SendAndClose(&common.Empty{}) diff --git a/controller/rpc/user_sync_epoch.go b/controller/rpc/user_sync_epoch.go new file mode 100644 index 0000000..bf6b72e --- /dev/null +++ b/controller/rpc/user_sync_epoch.go @@ -0,0 +1,17 @@ +package rpc + +import ( + "errors" + + "github.com/pasarguard/node/controller" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func userSyncError(err error) error { + var epochErr *controller.UserSyncEpochError + if errors.As(err, &epochErr) { + return status.Error(codes.FailedPrecondition, epochErr.Error()) + } + return err +} diff --git a/controller/rpc/user_sync_epoch_test.go b/controller/rpc/user_sync_epoch_test.go new file mode 100644 index 0000000..534de84 --- /dev/null +++ b/controller/rpc/user_sync_epoch_test.go @@ -0,0 +1,47 @@ +package rpc + +import ( + "context" + "testing" + + "github.com/google/uuid" + "github.com/pasarguard/node/common" + "github.com/pasarguard/node/config" + "github.com/pasarguard/node/controller" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func TestGRPCUserSyncEpochErrorUsesFailedPrecondition(t *testing.T) { + err := userSyncError(&controller.UserSyncEpochError{Received: 2, Current: 3}) + if status.Code(err) != codes.FailedPrecondition { + t.Fatalf("expected FailedPrecondition, got %v", status.Code(err)) + } +} + +func TestGRPCStaleUserSyncRejectsBeforeBackendAccess(t *testing.T) { + service := New(config.NewTestConfig(t.TempDir(), uuid.New())) + if err := service.ApplyUserSyncEpoch(70, func() error { return nil }); err != nil { + t.Fatalf("failed to establish current epoch: %v", err) + } + _, err := service.SyncUsers(context.Background(), &common.Users{UserSyncEpoch: 69}) + if status.Code(err) != codes.FailedPrecondition { + t.Fatalf("expected stale request to fail before nil backend access, got %v", err) + } +} + +func TestGRPCUserSyncStreamHasAggregatePayloadLimit(t *testing.T) { + user := &common.User{Email: "bounded@example.com"} + _, err := addUserSyncStreamPayload(common.MaxProtoBodyBytes-1, user) + if status.Code(err) != codes.ResourceExhausted { + t.Fatalf("expected aggregate stream limit, got %v", err) + } +} + +func TestGRPCChunkedUserSyncStreamHasAggregatePayloadLimit(t *testing.T) { + chunk := &common.UsersChunk{Users: []*common.User{{Email: "bounded@example.com"}}} + _, err := addUserSyncStreamPayload(common.MaxProtoBodyBytes-1, chunk) + if status.Code(err) != codes.ResourceExhausted { + t.Fatalf("expected aggregate chunk stream limit, got %v", err) + } +} diff --git a/controller/user_sync_epoch_test.go b/controller/user_sync_epoch_test.go new file mode 100644 index 0000000..04bd010 --- /dev/null +++ b/controller/user_sync_epoch_test.go @@ -0,0 +1,203 @@ +package controller + +import ( + "errors" + "reflect" + "sync" + "testing" + + "github.com/google/uuid" + "github.com/pasarguard/node/common" + "github.com/pasarguard/node/config" +) + +func newEpochTestController(t *testing.T) *Controller { + t.Helper() + return New(config.NewTestConfig(t.TempDir(), uuid.New())) +} + +func TestUserSyncEpochZeroAllowedUntilEpochAwareMutation(t *testing.T) { + c := newEpochTestController(t) + mutations := 0 + mutate := func() error { + mutations++ + return nil + } + + if err := c.ApplyUserSyncEpoch(0, mutate); err != nil { + t.Fatalf("legacy mutation before epoch rollout failed: %v", err) + } + if err := c.ApplyUserSyncEpoch(7, mutate); err != nil { + t.Fatalf("epoch-aware mutation failed: %v", err) + } + if err := c.ApplyUserSyncEpoch(0, mutate); err == nil { + t.Fatal("legacy mutation was accepted after epoch rollout") + } else { + var epochErr *UserSyncEpochError + if !errors.As(err, &epochErr) || epochErr.Current != 7 { + t.Fatalf("expected current epoch 7, got %v", err) + } + } + if mutations != 2 { + t.Fatalf("rejected mutation ran; got %d calls", mutations) + } +} + +func TestFailedHigherUserSyncEpochIsConsumed(t *testing.T) { + c := newEpochTestController(t) + wantErr := errors.New("backend failed") + if err := c.ApplyUserSyncEpoch(12, func() error { return wantErr }); !errors.Is(err, wantErr) { + t.Fatalf("expected backend failure, got %v", err) + } + + ran := false + err := c.ApplyUserSyncEpoch(11, func() error { + ran = true + return nil + }) + if err == nil { + t.Fatal("older mutation was accepted after failed higher epoch") + } + if ran { + t.Fatal("older backend mutation ran") + } +} + +func TestStaleLatePartialCannotOverwriteAuthoritativeSnapshot(t *testing.T) { + c := newEpochTestController(t) + state := []string{"initial"} + + if err := c.ApplyUserSyncEpoch(20, func() error { + state = []string{"authoritative"} + return nil + }); err != nil { + t.Fatalf("authoritative mutation failed: %v", err) + } + + err := c.ApplyUserSyncEpoch(19, func() error { + state = append(state, "stale-partial") + return nil + }) + if err == nil { + t.Fatal("stale late partial mutation was accepted") + } + if !reflect.DeepEqual(state, []string{"authoritative"}) { + t.Fatalf("stale mutation changed state: %v", state) + } +} + +func TestUserSyncEpochSerializesReservationAndMutation(t *testing.T) { + c := newEpochTestController(t) + firstStarted := make(chan struct{}) + releaseFirst := make(chan struct{}) + secondDone := make(chan struct{}) + secondAttempted := make(chan struct{}) + var orderMu sync.Mutex + order := make([]uint64, 0, 2) + + go func() { + _ = c.ApplyUserSyncEpoch(30, func() error { + close(firstStarted) + <-releaseFirst + orderMu.Lock() + order = append(order, 30) + orderMu.Unlock() + return nil + }) + }() + <-firstStarted + + go func() { + close(secondAttempted) + _ = c.ApplyUserSyncEpoch(31, func() error { + orderMu.Lock() + order = append(order, 31) + orderMu.Unlock() + return nil + }) + close(secondDone) + }() + <-secondAttempted + + select { + case <-secondDone: + t.Fatal("higher epoch mutated while the older mutation was still running") + default: + } + close(releaseFirst) + <-secondDone + + orderMu.Lock() + defer orderMu.Unlock() + if !reflect.DeepEqual(order, []uint64{30, 31}) { + t.Fatalf("mutations ran out of order: %v", order) + } +} + +func TestUserSyncEpochBatchRejectsMixedEpochs(t *testing.T) { + var batch UserSyncEpochBatch + if err := batch.Add(40); err != nil { + t.Fatalf("first epoch failed: %v", err) + } + if err := batch.Add(40); err != nil { + t.Fatalf("matching epoch failed: %v", err) + } + if err := batch.Add(41); err == nil { + t.Fatal("mixed epoch was accepted") + } + if batch.Epoch() != 40 { + t.Fatalf("mixed item changed batch epoch to %d", batch.Epoch()) + } +} + +func TestBaseInfoAdvertisesUserSyncEpochSupport(t *testing.T) { + c := newEpochTestController(t) + if !c.BaseInfoResponse().GetUserSyncEpochSupported() { + t.Fatal("base info did not advertise user sync epoch support") + } +} + +func TestBaseInfoReportsCurrentUserSyncEpochAfterSuccess(t *testing.T) { + c := newEpochTestController(t) + if err := c.ApplyUserSyncEpoch(60, func() error { return nil }); err != nil { + t.Fatalf("epoch mutation failed: %v", err) + } + if got := c.BaseInfoResponse().GetUserSyncEpoch(); got != 60 { + t.Fatalf("base info epoch = %d, want 60", got) + } +} + +func TestBaseInfoReportsConsumedUserSyncEpochAfterFailure(t *testing.T) { + c := newEpochTestController(t) + if err := c.ApplyUserSyncEpoch(61, func() error { return errors.New("backend failed") }); err == nil { + t.Fatal("expected backend failure") + } + if got := c.BaseInfoResponse().GetUserSyncEpoch(); got != 61 { + t.Fatalf("base info epoch = %d, want consumed epoch 61", got) + } +} + +func TestStaleStartIsRejectedBeforeDisconnect(t *testing.T) { + c := newEpochTestController(t) + shutdownStarted := make(chan struct{}) + allowShutdown := make(chan struct{}) + close(allowShutdown) + oldBackend := &blockingBackend{shutdownStarted: shutdownStarted, allowShutdown: allowShutdown} + c.backend = oldBackend + + if err := c.ApplyUserSyncEpoch(50, func() error { return nil }); err != nil { + t.Fatalf("failed to establish current epoch: %v", err) + } + err := c.StartBackendControlled(t.Context(), &common.Backend{UserSyncEpoch: 49}, "127.0.0.1") + if err == nil { + t.Fatal("stale start was accepted") + } + select { + case <-shutdownStarted: + t.Fatal("stale start disconnected the current backend") + default: + } + if c.Backend() != oldBackend { + t.Fatal("stale start replaced the current backend") + } +} diff --git a/docker-compose.yml b/docker-compose.yml index 3348aff..2f66929 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -20,8 +20,9 @@ services: SSL_CERT_FILE: "/var/lib/pg-node/certs/ssl_cert.pem" SSL_KEY_FILE: "/var/lib/pg-node/certs/ssl_key.pem" - # api key must be a valid uuid (you can use any version you want) - # API_KEY: xxxxxxxx-yyyy-zzzz-mmmm-aaaaaaaaaaa + # API_KEY must be a non-zero UUID. Docker Compose refuses to start if it + # is not supplied instead of exposing an API protected by uuid.Nil. + API_KEY: ${API_KEY:?set API_KEY to a non-zero UUID} GENERATED_CONFIG_PATH: "/var/lib/pg-node/generated" From a0aa9a915dfd1ee08421dc2af4a1f2e03f9add5b Mon Sep 17 00:00:00 2001 From: Rerowros Date: Mon, 10 Aug 2026 23:24:45 +0400 Subject: [PATCH 2/2] fix(node): bound lifecycle and user sync operations --- .env.example | 6 ++- README.md | 15 ++++++ backend/xray/config.go | 7 +++ backend/xray/user.go | 43 ++++++++------- backend/xray/user_removal_test.go | 12 ++++- backend/xray/user_runtime_consistency_test.go | 26 ++++++++++ controller/controller.go | 29 ++++++----- controller/controller_test.go | 35 +++++++++++++ controller/rest/base.go | 16 +++++- controller/rest/base_timeout_test.go | 52 +++++++++++++++++++ controller/rpc/base.go | 10 +++- controller/rpc/service.go | 19 ++++++- controller/rpc/user.go | 42 ++++++++++++--- controller/rpc/user_sync_epoch.go | 6 ++- controller/rpc/user_sync_epoch_test.go | 36 +++++++++++++ 15 files changed, 304 insertions(+), 50 deletions(-) create mode 100644 controller/rest/base_timeout_test.go diff --git a/.env.example b/.env.example index 5973438..1a2fafa 100644 --- a/.env.example +++ b/.env.example @@ -9,8 +9,9 @@ NODE_HOST = "0.0.0.0" SSL_CERT_FILE = /var/lib/pg-node/certs/ssl_cert.pem SSL_KEY_FILE = /var/lib/pg-node/certs/ssl_key.pem -# api key must be a valid uuid (you can use any version you want) -API_KEY = xxxxxxxx-yyyy-zzzz-mmmm-aaaaaaaaaaa +# Required: generate a non-zero UUID (for example, with `uuidgen`) and configure +# the same value in the panel. Placeholder and zero UUID values are rejected. +API_KEY = 01234567-89ab-4def-8123-456789abcdef ### can be rest or grpc # SERVICE_PROTOCOL = grpc @@ -20,6 +21,7 @@ API_KEY = xxxxxxxx-yyyy-zzzz-mmmm-aaaaaaaaaaa # GENERATED_CONFIG_PATH = /var/lib/pg-node/generated # LOG_BUFFER_SIZE = 10000 # STARTUP_LOG_TAIL_SIZE = 200 +# Both statistics intervals must be positive integers when overridden. # STATS_UPDATE_INTERVAL_SECONDS = 10 # STATS_CLEANUP_INTERVAL_SECONDS = 300 diff --git a/README.md b/README.md index b93efc7..8caceda 100644 --- a/README.md +++ b/README.md @@ -17,6 +17,21 @@ # Documentation You can find a full guide in docs https://docs.pasarguard.org/en/node/ +## Upgrade notes + +This release makes startup configuration validation strict: + +- `API_KEY` is required and must be a non-zero UUID. Generate a new value for + each node (for example, with `uuidgen`) and configure the same value in the + panel before restarting the node. The placeholder from older `.env` files is + not valid. +- `STATS_UPDATE_INTERVAL_SECONDS` and `STATS_CLEANUP_INTERVAL_SECONDS`, when + set, must be positive integers. Remove old zero or negative overrides to use + the defaults (`10` and `300` seconds), or replace them with positive values. + +The node now exits during startup with a configuration error instead of running +with an unauthenticated key or invalid statistics intervals. + # One-Click Installation (Recommended) The easiest way to install PasarGuard Node is using our automated installation script: diff --git a/backend/xray/config.go b/backend/xray/config.go index 8199a75..c5c3809 100644 --- a/backend/xray/config.go +++ b/backend/xray/config.go @@ -307,6 +307,13 @@ func (i *Inbound) removeUser(email string) { } } +func (i *Inbound) hasUser(email string) bool { + i.mu.RLock() + defer i.mu.RUnlock() + _, ok := i.clients[email] + return ok +} + type Stats struct{} func (c *Config) ToBytes() ([]byte, error) { diff --git a/backend/xray/user.go b/backend/xray/user.go index 26ed610..c460abf 100644 --- a/backend/xray/user.go +++ b/backend/xray/user.go @@ -61,27 +61,28 @@ func setupUserAccount(user *common.User) (api.ProxySettings, error) { // removeInboundUser treats an already-absent runtime user as a successful // idempotent revoke, but never hides transport/core errors. A caller must not // report a revoked credential while Xray still accepts it. -func removeInboundUser(ctx context.Context, handler inboundUserHandler, tag, email string) error { - err := handler.RemoveInboundUser(ctx, tag, email) - if isBenignUserRemovalError(err) { +func removeInboundUser(ctx context.Context, handler inboundUserHandler, inbound *Inbound, email string) error { + err := handler.RemoveInboundUser(ctx, inbound.Tag, email) + if isBenignUserRemovalError(err, inbound, email) { return nil } return err } -func isBenignUserRemovalError(err error) bool { +func isBenignUserRemovalError(err error, inbound *Inbound, email string) bool { if err == nil || status.Code(err) == codes.NotFound { return true } - // Xray's HandlerService historically reports both of these idempotent - // conditions as Unknown rather than NotFound. An absent user cannot keep a - // credential alive, and API/non-user-manager inbounds cannot contain one. - // All transport and runtime failures stay visible to the caller. - if status.Code(err) != codes.Unknown { + // Xray sometimes collapses an absent-user result into Unknown. Only accept + // that ambiguous status when it is a real gRPC status and the authoritative + // inbound snapshot already confirms that no credential can be resurrected. + // This deliberately avoids message matching: transport/core failures and an + // Unknown result for a cached user remain actionable errors. + grpcStatus, ok := status.FromError(err) + if !ok || grpcStatus.Code() != codes.Unknown || inbound == nil { return false } - message := strings.ToLower(status.Convert(err).Message()) - return strings.Contains(message, "not found") || strings.Contains(message, "not a usermanager") + return !inbound.hasUser(email) } func inboundFlow(inbound *Inbound) string { @@ -181,7 +182,7 @@ func (x *Xray) SyncUser(ctx context.Context, user *common.User) error { continue } - if err := removeInboundUser(ctx, handler, inbound.Tag, user.Email); err != nil { + if err := removeInboundUser(ctx, handler, inbound, user.Email); err != nil { return fmt.Errorf("failed to remove user %q from inbound %q: %w", user.Email, inbound.Tag, err) } // Keep the restart snapshot aligned with every confirmed runtime @@ -266,7 +267,7 @@ func (x *Xray) UpdateUsers(ctx context.Context, users []*common.User) error { handler := x.inboundUserHandler() inboundByTag, updates := x.config.buildInboundUpdates(users) - var errMessage strings.Builder + var updateErrors []error for tag, update := range updates { removeEmails := make([]string, 0, len(update.removeEmailSet)) @@ -278,29 +279,31 @@ func (x *Xray) UpdateUsers(ctx context.Context, users []*common.User) error { inbound := inboundByTag[tag] for _, email := range removeEmails { - if err := removeInboundUser(ctx, handler, tag, email); err != nil { - return fmt.Errorf("failed to remove user %q from inbound %q: %w", email, tag, err) + if err := removeInboundUser(ctx, handler, inbound, email); err != nil { + updateErrors = append(updateErrors, fmt.Errorf("failed to remove user %q from inbound %q: %w", email, tag, err)) + continue } inbound.removeUser(email) } for _, account := range update.accounts { email := account.GetEmail() - if err := removeInboundUser(ctx, handler, tag, email); err != nil { - return fmt.Errorf("failed to replace user %q in inbound %q: %w", email, tag, err) + if err := removeInboundUser(ctx, handler, inbound, email); err != nil { + updateErrors = append(updateErrors, fmt.Errorf("failed to replace user %q in inbound %q: %w", email, tag, err)) + continue } inbound.removeUser(email) if err := handler.AddInboundUser(ctx, tag, accountForAPI(inbound, account)); err != nil { log.Println(err) - errMessage.WriteString("\n" + err.Error()) + updateErrors = append(updateErrors, fmt.Errorf("failed to add user %q to inbound %q: %w", email, tag, err)) continue } inbound.updateUser(account) } } - if errMessage.String() != "" { - return errors.New("failed to update users:" + errMessage.String()) + if len(updateErrors) > 0 { + return fmt.Errorf("failed to update users: %w", errors.Join(updateErrors...)) } return nil diff --git a/backend/xray/user_removal_test.go b/backend/xray/user_removal_test.go index 0ed3dac..a0e3725 100644 --- a/backend/xray/user_removal_test.go +++ b/backend/xray/user_removal_test.go @@ -4,28 +4,36 @@ import ( "errors" "testing" + "github.com/pasarguard/node/backend/xray/api" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) func TestIsBenignUserRemovalError(t *testing.T) { + absentInbound := &Inbound{clients: make(map[string]api.Account)} for _, err := range []error{ nil, status.Error(codes.NotFound, "user not found"), status.Error(codes.Unknown, "proxy/trojan: User user@example.com not found."), status.Error(codes.Unknown, "app/proxyman/command: proxy is not a UserManager"), + status.Error(codes.Unknown, "opaque handler response"), } { - if !isBenignUserRemovalError(err) { + if !isBenignUserRemovalError(err, absentInbound, "user@example.com") { t.Fatalf("expected benign removal error: %v", err) } } + presentInbound := &Inbound{clients: map[string]api.Account{ + "user@example.com": trojanAccount("user@example.com", "secret"), + }} for _, err := range []error{ + status.Error(codes.Unknown, "user not found"), status.Error(codes.Unavailable, "connection refused"), status.Error(codes.DeadlineExceeded, "deadline exceeded"), + errors.New("user not found"), errors.New("local handler failure"), } { - if isBenignUserRemovalError(err) { + if isBenignUserRemovalError(err, presentInbound, "user@example.com") { t.Fatalf("unexpectedly accepted runtime failure: %v", err) } } diff --git a/backend/xray/user_runtime_consistency_test.go b/backend/xray/user_runtime_consistency_test.go index 800f068..fd07b49 100644 --- a/backend/xray/user_runtime_consistency_test.go +++ b/backend/xray/user_runtime_consistency_test.go @@ -256,3 +256,29 @@ func TestSyncUserAddFailureDoesNotResurrectRemovedCredential(t *testing.T) { t.Fatalf("restart snapshot would resurrect removed credential: %#v", snapshot) } } + +func TestUpdateUsersAggregatesRemovalFailuresAndContinues(t *testing.T) { + oldA := trojanAccount("a@example.com", "old-a") + oldB := trojanAccount("b@example.com", "old-b") + oldC := trojanAccount("c@example.com", "old-c") + x, inbound, handler := newRuntimeConsistencyXray(oldA, oldB, oldC) + firstFailure := errors.New("first remove failed") + secondFailure := errors.New("second remove failed") + handler.removeFailures[1] = firstFailure + handler.removeFailures[2] = secondFailure + + err := x.UpdateUsers(context.Background(), []*common.User{ + trojanUser("a@example.com", "unused"), + trojanUser("b@example.com", "unused"), + trojanUser("c@example.com", "unused"), + }) + if err == nil || !errors.Is(err, firstFailure) || !errors.Is(err, secondFailure) { + t.Fatalf("expected both removal errors to be aggregated, got %v", err) + } + if handler.removeCalls != 3 { + t.Fatalf("remove calls = %d, want 3", handler.removeCalls) + } + if _, ok := accountPassword(t, inbound.clients, "c@example.com"); ok { + t.Fatal("later successful removal remained in restart snapshot") + } +} diff --git a/controller/controller.go b/controller/controller.go index 6dec7d9..2926d11 100644 --- a/controller/controller.go +++ b/controller/controller.go @@ -6,6 +6,7 @@ import ( "fmt" "log" "sync" + "sync/atomic" "time" "github.com/google/uuid" @@ -38,7 +39,7 @@ type Controller struct { mu sync.RWMutex controlMu sync.Mutex userSyncMu sync.Mutex - maxUserSyncEpoch uint64 + maxUserSyncEpoch *atomic.Uint64 } // UserSyncEpochError reports a user mutation that is older than one already @@ -79,10 +80,11 @@ func (e *UserSyncEpochError) Error() string { func New(cfg *config.Config) *Controller { _, cancel := context.WithCancel(context.Background()) return &Controller{ - cfg: cfg, - apiPort: netutil.FindFreePort(), - metricPort: netutil.FindFreePort(), - cancelFunc: cancel, + cfg: cfg, + apiPort: netutil.FindFreePort(), + metricPort: netutil.FindFreePort(), + cancelFunc: cancel, + maxUserSyncEpoch: &atomic.Uint64{}, } } @@ -165,20 +167,19 @@ func (c *Controller) ApplyUserSyncEpoch(epoch uint64, mutate func() error) error c.userSyncMu.Lock() defer c.userSyncMu.Unlock() - if epoch < c.maxUserSyncEpoch || (epoch == 0 && c.maxUserSyncEpoch > 0) { - return &UserSyncEpochError{Received: epoch, Current: c.maxUserSyncEpoch} + currentEpoch := c.maxUserSyncEpoch.Load() + if epoch < currentEpoch || (epoch == 0 && currentEpoch > 0) { + return &UserSyncEpochError{Received: epoch, Current: currentEpoch} } - if epoch > c.maxUserSyncEpoch { - c.maxUserSyncEpoch = epoch + if epoch > currentEpoch { + c.maxUserSyncEpoch.Store(epoch) } return mutate() } func (c *Controller) UserSyncEpoch() uint64 { - c.userSyncMu.Lock() - defer c.userSyncMu.Unlock() - return c.maxUserSyncEpoch + return c.maxUserSyncEpoch.Load() } func (c *Controller) NewRequest() { @@ -349,8 +350,8 @@ func (c *Controller) SystemStats(ctx context.Context) *common.SystemStatsRespons func (c *Controller) BaseInfoResponse() *common.BaseInfoResponse { userSyncEpoch := c.UserSyncEpoch() - c.mu.Lock() - defer c.mu.Unlock() + c.mu.RLock() + defer c.mu.RUnlock() response := &common.BaseInfoResponse{ Started: false, diff --git a/controller/controller_test.go b/controller/controller_test.go index 77ed04a..93ad78a 100644 --- a/controller/controller_test.go +++ b/controller/controller_test.go @@ -4,6 +4,7 @@ import ( "testing" "time" + "github.com/pasarguard/node/common" "github.com/pasarguard/node/config" ) @@ -29,3 +30,37 @@ func TestStaleKeepAliveCannotDisconnectReplacementConnection(t *testing.T) { t.Fatal("stale keep-alive generation became valid after replacement") } } + +func TestBaseInfoDoesNotWaitForUserMutation(t *testing.T) { + c := New(&config.Config{}) + mutationStarted := make(chan struct{}) + releaseMutation := make(chan struct{}) + mutationDone := make(chan struct{}) + go func() { + _ = c.ApplyUserSyncEpoch(42, func() error { + close(mutationStarted) + <-releaseMutation + return nil + }) + close(mutationDone) + }() + <-mutationStarted + + responseDone := make(chan *common.BaseInfoResponse, 1) + go func() { responseDone <- c.BaseInfoResponse() }() + select { + case response := <-responseDone: + if response.GetUserSyncEpoch() != 42 { + t.Fatalf("BaseInfo epoch = %d, want 42", response.GetUserSyncEpoch()) + } + case <-time.After(100 * time.Millisecond): + t.Fatal("BaseInfo blocked behind an in-flight user mutation") + } + + close(releaseMutation) + select { + case <-mutationDone: + case <-time.After(time.Second): + t.Fatal("mutation did not finish") + } +} diff --git a/controller/rest/base.go b/controller/rest/base.go index 26fd864..54e915c 100644 --- a/controller/rest/base.go +++ b/controller/rest/base.go @@ -12,6 +12,13 @@ func (s *Service) Base(w http.ResponseWriter, _ *http.Request) { } func (s *Service) Start(w http.ResponseWriter, r *http.Request) { + if err := disableWriteDeadline(w); err != nil && !errors.Is(err, http.ErrNotSupported) { + http.Error(w, "failed to configure lifecycle response deadline", http.StatusInternalServerError) + return + } + stopCancelDeadline := stopWritesOnContext(r.Context(), w) + defer stopCancelDeadline() + s.LockControl() defer s.UnlockControl() @@ -45,7 +52,14 @@ func (s *Service) Start(w http.ResponseWriter, r *http.Request) { common.SendProtoResponse(w, s.BaseInfoResponse()) } -func (s *Service) Stop(w http.ResponseWriter, _ *http.Request) { +func (s *Service) Stop(w http.ResponseWriter, r *http.Request) { + if err := disableWriteDeadline(w); err != nil && !errors.Is(err, http.ErrNotSupported) { + http.Error(w, "failed to configure lifecycle response deadline", http.StatusInternalServerError) + return + } + stopCancelDeadline := stopWritesOnContext(r.Context(), w) + defer stopCancelDeadline() + s.LockControl() defer s.UnlockControl() diff --git a/controller/rest/base_timeout_test.go b/controller/rest/base_timeout_test.go new file mode 100644 index 0000000..a834cbd --- /dev/null +++ b/controller/rest/base_timeout_test.go @@ -0,0 +1,52 @@ +package rest + +import ( + "bytes" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/google/uuid" + "github.com/pasarguard/node/config" +) + +type deadlineRecorder struct { + *httptest.ResponseRecorder + deadlines []time.Time +} + +func (w *deadlineRecorder) SetWriteDeadline(deadline time.Time) error { + w.deadlines = append(w.deadlines, deadline) + return nil +} + +func TestStopDisablesGlobalWriteTimeout(t *testing.T) { + service := New(config.NewTestConfig(t.TempDir(), uuid.New())) + w := &deadlineRecorder{ResponseRecorder: httptest.NewRecorder()} + r := httptest.NewRequest(http.MethodPut, "/stop", nil) + + service.Stop(w, r) + + if len(w.deadlines) == 0 || !w.deadlines[0].IsZero() { + t.Fatalf("first write deadline = %v, want disabled deadline", w.deadlines) + } + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", w.Code) + } +} + +func TestStartDisablesGlobalWriteTimeoutBeforeReadingRequest(t *testing.T) { + service := New(config.NewTestConfig(t.TempDir(), uuid.New())) + w := &deadlineRecorder{ResponseRecorder: httptest.NewRecorder()} + r := httptest.NewRequest(http.MethodPost, "/start", bytes.NewReader([]byte{0xff})) + + service.Start(w, r) + + if len(w.deadlines) == 0 || !w.deadlines[0].IsZero() { + t.Fatalf("first write deadline = %v, want disabled deadline", w.deadlines) + } + if w.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400 for empty protobuf body", w.Code) + } +} diff --git a/controller/rpc/base.go b/controller/rpc/base.go index 1fb2ef6..cda3f92 100644 --- a/controller/rpc/base.go +++ b/controller/rpc/base.go @@ -2,6 +2,7 @@ package rpc import ( "context" + "log" "github.com/pasarguard/node/common" "google.golang.org/grpc/codes" @@ -22,7 +23,14 @@ func (s *Service) Start(ctx context.Context, data *common.Backend) (*common.Base } if err := s.StartBackendControlled(ctx, data, clientIP); err != nil { - return nil, userSyncError(err) + if epochErr := userSyncError(err); status.Code(epochErr) == codes.FailedPrecondition { + return nil, epochErr + } + if ctx.Err() != nil { + return nil, status.FromContextError(ctx.Err()).Err() + } + log.Print("backend start failed") + return nil, status.Error(codes.Internal, "failed to start backend") } return s.BaseInfoResponse(), nil diff --git a/controller/rpc/service.go b/controller/rpc/service.go index d633c41..df8e74a 100644 --- a/controller/rpc/service.go +++ b/controller/rpc/service.go @@ -11,17 +11,34 @@ import ( "github.com/pasarguard/node/config" "github.com/pasarguard/node/controller" "google.golang.org/grpc" + "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials" + "google.golang.org/grpc/status" ) type Service struct { common.UnimplementedNodeServiceServer controller.Controller + bufferedUserSyncs chan struct{} } +const maxConcurrentBufferedUserSyncs = 2 + func New(cfg *config.Config) *Service { return &Service{ - Controller: *controller.New(cfg), + Controller: *controller.New(cfg), + bufferedUserSyncs: make(chan struct{}, maxConcurrentBufferedUserSyncs), + } +} + +func (s *Service) acquireBufferedUserSync(ctx context.Context) (func(), error) { + select { + case s.bufferedUserSyncs <- struct{}{}: + return func() { <-s.bufferedUserSyncs }, nil + case <-ctx.Done(): + return nil, status.FromContextError(ctx.Err()).Err() + default: + return nil, status.Error(codes.ResourceExhausted, "too many buffered user sync streams") } } diff --git a/controller/rpc/user.go b/controller/rpc/user.go index 4bf1f1c..4d0afaf 100644 --- a/controller/rpc/user.go +++ b/controller/rpc/user.go @@ -24,6 +24,12 @@ func addUserSyncStreamPayload(total int64, message proto.Message) (int64, error) } func (s *Service) SyncUser(stream grpc.ClientStreamingServer[common.User, common.Empty]) error { + release, err := s.acquireBufferedUserSync(stream.Context()) + if err != nil { + return err + } + defer release() + users := make([]*common.User, 0) var epochBatch controller.UserSyncEpochBatch var streamBytes int64 @@ -34,11 +40,11 @@ func (s *Service) SyncUser(stream grpc.ClientStreamingServer[common.User, common break } if err != nil { - return status.Errorf(codes.Internal, "failed to receive user: %v", err) + return sanitizedStreamReceiveError(err, "failed to receive user stream") } if user.GetEmail() == "" { - return errors.New("email is required") + return status.Error(codes.InvalidArgument, "email is required") } streamBytes, err = addUserSyncStreamPayload(streamBytes, user) if err != nil { @@ -48,7 +54,6 @@ func (s *Service) SyncUser(stream grpc.ClientStreamingServer[common.User, common return status.Error(codes.InvalidArgument, err.Error()) } - log.Printf("Got user: %v", user.GetEmail()) users = append(users, user) } @@ -63,8 +68,8 @@ func (s *Service) SyncUser(stream grpc.ClientStreamingServer[common.User, common } for _, user := range users { if err = back.SyncUser(stream.Context(), user); err != nil { - log.Printf("Error syncing user: %v", err) - return status.Errorf(codes.Internal, "failed to update user: %v", err) + log.Printf("user sync backend mutation failed") + return status.Error(codes.Internal, "failed to update user") } } return nil @@ -81,7 +86,11 @@ func (s *Service) SyncUsers(ctx context.Context, users *common.Users) (*common.E if err != nil { return err } - return back.SyncUsers(ctx, users.GetUsers()) + if err = back.SyncUsers(ctx, users.GetUsers()); err != nil { + log.Printf("bulk user sync backend mutation failed") + return status.Error(codes.Internal, "failed to update users") + } + return nil }); err != nil { return nil, userSyncError(err) } @@ -90,6 +99,12 @@ func (s *Service) SyncUsers(ctx context.Context, users *common.Users) (*common.E } func (s *Service) SyncUsersChunked(stream grpc.ClientStreamingServer[common.UsersChunk, common.Empty]) error { + release, err := s.acquireBufferedUserSync(stream.Context()) + if err != nil { + return err + } + defer release() + chunks := make(map[uint64][]*common.User) var ( lastIndex uint64 @@ -104,7 +119,7 @@ func (s *Service) SyncUsersChunked(stream grpc.ClientStreamingServer[common.User break } if err != nil { - return status.Errorf(codes.Internal, "failed to receive chunk: %v", err) + return sanitizedStreamReceiveError(err, "failed to receive user chunk stream") } streamBytes, err = addUserSyncStreamPayload(streamBytes, chunk) if err != nil { @@ -134,7 +149,8 @@ func (s *Service) SyncUsersChunked(stream grpc.ClientStreamingServer[common.User return err } if err = controller.ApplyChunkedUserUpdate(stream.Context(), back, users); err != nil { - return status.Errorf(codes.Internal, "failed to update users: %v", err) + log.Printf("chunked user sync backend mutation failed") + return status.Error(codes.Internal, "failed to update users") } return nil }); err != nil { @@ -143,3 +159,13 @@ func (s *Service) SyncUsersChunked(stream grpc.ClientStreamingServer[common.User return stream.SendAndClose(&common.Empty{}) } + +func sanitizedStreamReceiveError(err error, message string) error { + if code := status.Code(err); code != codes.Unknown { + return status.Error(code, message) + } + if contextErr := status.FromContextError(err); contextErr.Code() != codes.Unknown { + return status.Error(contextErr.Code(), message) + } + return status.Error(codes.Internal, message) +} diff --git a/controller/rpc/user_sync_epoch.go b/controller/rpc/user_sync_epoch.go index bf6b72e..d3427d3 100644 --- a/controller/rpc/user_sync_epoch.go +++ b/controller/rpc/user_sync_epoch.go @@ -13,5 +13,9 @@ func userSyncError(err error) error { if errors.As(err, &epochErr) { return status.Error(codes.FailedPrecondition, epochErr.Error()) } - return err + code := status.Code(err) + if code == codes.Unknown { + code = codes.Internal + } + return status.Error(code, "user sync failed") } diff --git a/controller/rpc/user_sync_epoch_test.go b/controller/rpc/user_sync_epoch_test.go index 534de84..e41dde4 100644 --- a/controller/rpc/user_sync_epoch_test.go +++ b/controller/rpc/user_sync_epoch_test.go @@ -2,6 +2,8 @@ package rpc import ( "context" + "errors" + "strings" "testing" "github.com/google/uuid" @@ -19,6 +21,40 @@ func TestGRPCUserSyncEpochErrorUsesFailedPrecondition(t *testing.T) { } } +func TestGRPCUserSyncErrorUsesTypedStatusWithoutPII(t *testing.T) { + err := userSyncError(errors.New("failed for private@example.com")) + if status.Code(err) != codes.Internal { + t.Fatalf("expected Internal, got %v", status.Code(err)) + } + if strings.Contains(err.Error(), "private@example.com") { + t.Fatalf("gRPC error leaked user identity: %v", err) + } +} + +func TestBufferedUserSyncConcurrencyIsBounded(t *testing.T) { + service := New(config.NewTestConfig(t.TempDir(), uuid.New())) + releases := make([]func(), 0, maxConcurrentBufferedUserSyncs) + for range maxConcurrentBufferedUserSyncs { + release, err := service.acquireBufferedUserSync(context.Background()) + if err != nil { + t.Fatalf("failed to acquire permitted slot: %v", err) + } + releases = append(releases, release) + } + if _, err := service.acquireBufferedUserSync(context.Background()); status.Code(err) != codes.ResourceExhausted { + t.Fatalf("excess stream status = %v, want ResourceExhausted", status.Code(err)) + } + releases[0]() + release, err := service.acquireBufferedUserSync(context.Background()) + if err != nil { + t.Fatalf("released slot was not reusable: %v", err) + } + release() + for _, release := range releases[1:] { + release() + } +} + func TestGRPCStaleUserSyncRejectsBeforeBackendAccess(t *testing.T) { service := New(config.NewTestConfig(t.TempDir(), uuid.New())) if err := service.ApplyUserSyncEpoch(70, func() error { return nil }); err != nil {