diff --git a/core/protocol/anytls/inbound.go b/core/protocol/anytls/inbound.go index f84395c7..69561fe7 100644 --- a/core/protocol/anytls/inbound.go +++ b/core/protocol/anytls/inbound.go @@ -5,10 +5,15 @@ // tearing the listener down (see core/inbound_users.go). // // UPGRADING sing-box: re-copy this file from the new tag and re-apply the -// change below. Nothing here will fail to compile if you forget, it will just +// changes below. Nothing here will fail to compile if you forget, it will just // silently keep running the old implementation. // -// Local change vs sing-box: none, other than the package clause. +// Local change vs sing-box: two lines, both for the user-session registry. +// NewInbound wraps the router with it, and NewConnection -- which owns an +// anytls session for as long as that session lives -- is redirected through +// newSessionConnection in users.go, so the session can be registered there and +// closed later. Without it, removing a user only stops new sessions while the +// one the client already holds keeps being served. See core/usersession. package anytls import ( @@ -54,6 +59,7 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo router: uot.NewRouter(router, logger), logger: logger, } + inbound.router = withUserSessions(inbound.router) if options.TLS != nil && options.TLS.Enabled { tlsConfig, err := tls.NewServer(ctx, logger, common.PtrValueOrDefault(options.TLS)) @@ -117,7 +123,7 @@ func (h *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata ada } conn = tlsConn } - err := h.service.NewConnection(adapter.WithContext(ctx, &metadata), conn, metadata.Source, onClose) + err := h.newSessionConnection(adapter.WithContext(ctx, &metadata), conn, metadata.Source, onClose) if err != nil { N.CloseOnHandshakeFailure(conn, onClose, err) h.logger.ErrorContext(ctx, E.Cause(err, "process connection from ", metadata.Source)) diff --git a/core/protocol/anytls/users.go b/core/protocol/anytls/users.go index 289319d1..07fa8b4e 100644 --- a/core/protocol/anytls/users.go +++ b/core/protocol/anytls/users.go @@ -1,15 +1,67 @@ package anytls import ( + "context" + "net" + + "github.com/shenaba/2s-ui/core/usersession" + + "github.com/sagernet/sing-box/adapter" "github.com/sagernet/sing-box/option" "github.com/sagernet/sing/common" + M "github.com/sagernet/sing/common/metadata" + N "github.com/sagernet/sing/common/network" anytls "github.com/anytls/sing-anytls" ) +// UpdateUsers swaps the user table of a running inbound, and cuts the sessions +// of everyone who just left it. Both halves are needed: the table alone only +// decides who may open a *new* session, while an anytls client that +// authenticated before the change keeps opening streams on the one it has. func (h *Inbound) UpdateUsers(users []option.AnyTLSUser) error { h.service.UpdateUsers(common.Map(users, func(it option.AnyTLSUser) anytls.User { return (anytls.User)(it) })) + // Unclosable is deliberately not checked: newSessionConnection registers + // the transport of every anytls session, so a removed user's session is + // always cut outright and never costs the inbound a rebuild. The only thing + // that can be counted here is a straggler stream that reached the router + // after its own session had already ended -- not a session, and not worth + // disconnecting everyone else for. If anytls ever grows a session this + // inbound does not hold the transport of, that stops being true. + h.sessions().CloseUsers(usersession.KeepSet(common.Map(users, func(it option.AnyTLSUser) string { + return it.Name + }))) return nil } + +// withUserSessions installs the session registry. anytls only needs the router +// to record who is connected -- the session itself is reached in +// newSessionConnection below, which is where it can be closed. +func withUserSessions(router adapter.ConnectionRouterEx) adapter.ConnectionRouterEx { + return usersession.WrapRouterEx(router, usersession.BindOnly) +} + +// sessions reaches the registry installed by withUserSessions. NewInbound wraps +// the router right after building the struct and nothing replaces the field +// afterwards, so this holds. It is a bare assertion on purpose: someone +// reordering that would otherwise turn every session hook here into a silent +// no-op, and a panic on the save path is the lesser outcome. +func (h *Inbound) sessions() *usersession.Registry { + return h.router.(*usersession.RouterEx).Registry() +} + +// newSessionConnection stands in for h.service.NewConnection at the one call +// site that owns an anytls session for its whole life. Registering the session +// here is what makes it closable; the streams it later opens reach the router +// individually and cannot be used to find it. +// +// The call blocks until the session ends, so the deferred Untrack is what keeps +// the registry to sessions that actually exist. +func (h *Inbound) newSessionConnection(ctx context.Context, conn net.Conn, source M.Socksaddr, onClose N.CloseHandlerFunc) error { + key := source.String() + h.sessions().Track(key, conn) + defer h.sessions().Untrack(key) + return h.service.NewConnection(ctx, conn, source, onClose) +} diff --git a/core/protocol/hysteria/inbound.go b/core/protocol/hysteria/inbound.go index f0c35c07..0a8c5ab8 100644 --- a/core/protocol/hysteria/inbound.go +++ b/core/protocol/hysteria/inbound.go @@ -5,7 +5,7 @@ // tearing the listener down (see core/inbound_users.go). // // UPGRADING sing-box: re-copy this file from the new tag and re-apply the -// change below. Nothing here will fail to compile if you forget, it will just +// changes below. Nothing here will fail to compile if you forget, it will just // silently keep running the old implementation. // // Local change vs sing-box: the service is keyed by user name @@ -14,6 +14,15 @@ // UpdateUsers does rewrite it, under live sessions, and a position is not. // Deleting a user shifted every later one, which mis-attributed traffic and // could index past the end of the name slice outright (upstream #1231). +// +// Second local change: one line installs the user-session registry +// +// inbound.router = withUserSessions(inbound.router) +// +// which is what lets an already authenticated session be reached when its user +// is removed from the inbound -- swapping the user table alone only decides who +// may start a new one. That line is the whole of it here; everything it reaches +// lives in users.go, which the copy check skips. See core/usersession. package hysteria import ( @@ -68,6 +77,7 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo }), tlsConfig: tlsConfig, } + inbound.router = withUserSessions(inbound.router) var sendBps, receiveBps uint64 if options.Up.Value() > 0 { sendBps = options.Up.Value() diff --git a/core/protocol/hysteria/users.go b/core/protocol/hysteria/users.go index b36b4711..3a56064c 100644 --- a/core/protocol/hysteria/users.go +++ b/core/protocol/hysteria/users.go @@ -1,9 +1,16 @@ package hysteria import ( + "github.com/shenaba/2s-ui/core/usersession" + + "github.com/sagernet/sing-box/adapter" "github.com/sagernet/sing-box/option" ) +// UpdateUsers swaps the user table of a running inbound and disconnects +// everyone who just left it -- see the note in tuic/users.go, hysteria reuses +// an authenticated session the same way, and a removed user who is still +// connected costs the inbound a rebuild for the same reason. func (h *Inbound) UpdateUsers(users []option.HysteriaUser) error { userList := make([]string, 0, len(users)) userPasswordList := make([]string, 0, len(users)) @@ -18,5 +25,15 @@ func (h *Inbound) UpdateUsers(users []option.HysteriaUser) error { userPasswordList = append(userPasswordList, password) } h.service.UpdateUsers(userList, userPasswordList) - return nil + return h.sessions().CloseUsers(usersession.KeepSet(userList)).RestartRequired() +} + +// withUserSessions installs the session registry in front of the router. This +// inbound holds the full adapter.Router, so it takes the Router shim. +func withUserSessions(router adapter.Router) adapter.Router { + return usersession.WrapRouter(router, usersession.BindOnly) +} + +func (h *Inbound) sessions() *usersession.Registry { + return h.router.(*usersession.Router).Registry() } diff --git a/core/protocol/hysteria2/inbound.go b/core/protocol/hysteria2/inbound.go index 416d33e9..d1741787 100644 --- a/core/protocol/hysteria2/inbound.go +++ b/core/protocol/hysteria2/inbound.go @@ -5,7 +5,7 @@ // tearing the listener down (see core/inbound_users.go). // // UPGRADING sing-box: re-copy this file from the new tag and re-apply the -// change below. Nothing here will fail to compile if you forget, it will just +// changes below. Nothing here will fail to compile if you forget, it will just // silently keep running the old implementation. // // Local change vs sing-box: the service is keyed by user name @@ -14,6 +14,15 @@ // UpdateUsers does rewrite it, under live sessions, and a position is not. // Deleting a user shifted every later one, which mis-attributed traffic and // could index past the end of the name slice outright (upstream #1231). +// +// Second local change: one line installs the user-session registry +// +// inbound.router = withUserSessions(inbound.router) +// +// which is what lets an already authenticated session be reached when its user +// is removed from the inbound -- swapping the user table alone only decides who +// may start a new one. That line is the whole of it here; everything it reaches +// lives in users.go, which the copy check skips. See core/usersession. package hysteria2 import ( @@ -139,6 +148,7 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo }), tlsConfig: tlsConfig, } + inbound.router = withUserSessions(inbound.router) var udpTimeout time.Duration if options.UDPTimeout != 0 { udpTimeout = time.Duration(options.UDPTimeout) diff --git a/core/protocol/hysteria2/users.go b/core/protocol/hysteria2/users.go index 2038dc8e..6b47ea09 100644 --- a/core/protocol/hysteria2/users.go +++ b/core/protocol/hysteria2/users.go @@ -1,9 +1,16 @@ package hysteria2 import ( + "github.com/shenaba/2s-ui/core/usersession" + + "github.com/sagernet/sing-box/adapter" "github.com/sagernet/sing-box/option" ) +// UpdateUsers swaps the user table of a running inbound and disconnects +// everyone who just left it -- see the note in tuic/users.go, hysteria2 reuses +// an authenticated session the same way, and a removed user who is still +// connected costs the inbound a rebuild for the same reason. func (h *Inbound) UpdateUsers(users []option.Hysteria2User) error { userList := make([]string, 0, len(users)) userPasswordList := make([]string, 0, len(users)) @@ -12,5 +19,15 @@ func (h *Inbound) UpdateUsers(users []option.Hysteria2User) error { userPasswordList = append(userPasswordList, user.Password) } h.service.UpdateUsers(userList, userPasswordList) - return nil + return h.sessions().CloseUsers(usersession.KeepSet(userList)).RestartRequired() +} + +// withUserSessions installs the session registry in front of the router. This +// inbound holds the full adapter.Router, so it takes the Router shim. +func withUserSessions(router adapter.Router) adapter.Router { + return usersession.WrapRouter(router, usersession.BindOnly) +} + +func (h *Inbound) sessions() *usersession.Registry { + return h.router.(*usersession.Router).Registry() } diff --git a/core/protocol/trojan/inbound.go b/core/protocol/trojan/inbound.go index 5dfb7077..67cad54d 100644 --- a/core/protocol/trojan/inbound.go +++ b/core/protocol/trojan/inbound.go @@ -5,7 +5,7 @@ // tearing the listener down (see core/inbound_users.go). // // UPGRADING sing-box: re-copy this file from the new tag and re-apply the -// change below. Nothing here will fail to compile if you forget, it will just +// changes below. Nothing here will fail to compile if you forget, it will just // silently keep running the old implementation. // // Local change vs sing-box: the service is keyed by user name @@ -14,6 +14,15 @@ // UpdateUsers does rewrite it, under live sessions, and a position is not. // Deleting a user shifted every later one, which mis-attributed traffic and // could index past the end of the name slice outright (upstream #1231). +// +// Second local change: one line installs the user-session registry +// +// inbound.router = withUserSessions(inbound.router) +// +// which is what lets an already authenticated session be reached when its user +// is removed from the inbound -- swapping the user table alone only decides who +// may start a new one. That line is the whole of it here; everything it reaches +// lives in users.go, which the copy check skips. See core/usersession. package trojan import ( @@ -118,6 +127,7 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo if err != nil { return nil, err } + inbound.router = withUserSessions(inbound.router) inbound.service = service inbound.listener = listener.New(listener.Options{ Context: ctx, diff --git a/core/protocol/trojan/users.go b/core/protocol/trojan/users.go index a90e1879..a2ce7d47 100644 --- a/core/protocol/trojan/users.go +++ b/core/protocol/trojan/users.go @@ -1,14 +1,43 @@ package trojan import ( + "github.com/shenaba/2s-ui/core/usersession" + + "github.com/sagernet/sing-box/adapter" "github.com/sagernet/sing-box/option" "github.com/sagernet/sing/common" ) +// UpdateUsers swaps the user table of a running inbound and cuts the multiplex +// sessions of everyone who just left it -- see the note in vless/users.go. +// Sessions are only cut once the table actually changed: trojan rejects a +// duplicate name and fails the whole update, and that must leave the inbound +// exactly as it was. func (h *Inbound) UpdateUsers(users []option.TrojanUser) error { - return h.service.UpdateUsers(common.Map(users, func(it option.TrojanUser) string { + err := h.service.UpdateUsers(common.Map(users, func(it option.TrojanUser) string { return it.Name }), common.Map(users, func(it option.TrojanUser) string { return it.Password })) + if err != nil { + return err + } + // Unclosable is not checked here; see the note in vless/users.go. + h.sessions().CloseUsers(usersession.KeepSet(common.Map(users, func(it option.TrojanUser) string { + return it.Name + }))) + return nil +} + +// withUserSessions installs the session registry in front of the router. Only +// the multiplex carrier is tracked: every other connection authenticates on its +// own and is already covered by ConnTracker. The fallback path routes +// unauthenticated visitors through this same router, which is one more reason +// nothing here may refuse a connection. +func withUserSessions(router adapter.ConnectionRouterEx) adapter.ConnectionRouterEx { + return usersession.WrapRouterEx(router, usersession.TrackMuxCarrier) +} + +func (h *Inbound) sessions() *usersession.Registry { + return h.router.(*usersession.RouterEx).Registry() } diff --git a/core/protocol/tuic/inbound.go b/core/protocol/tuic/inbound.go index 668d2e12..d2431ef8 100644 --- a/core/protocol/tuic/inbound.go +++ b/core/protocol/tuic/inbound.go @@ -5,7 +5,7 @@ // tearing the listener down (see core/inbound_users.go). // // UPGRADING sing-box: re-copy this file from the new tag and re-apply the -// change below. Nothing here will fail to compile if you forget, it will just +// changes below. Nothing here will fail to compile if you forget, it will just // silently keep running the old implementation. // // Local change vs sing-box: the service is keyed by user name @@ -14,6 +14,15 @@ // UpdateUsers does rewrite it, under live sessions, and a position is not. // Deleting a user shifted every later one, which mis-attributed traffic and // could index past the end of the name slice outright (upstream #1231). +// +// Second local change: one line installs the user-session registry +// +// inbound.router = withUserSessions(inbound.router) +// +// which is what lets an already authenticated session be reached when its user +// is removed from the inbound -- swapping the user table alone only decides who +// may start a new one. That line is the whole of it here; everything it reaches +// lives in users.go, which the copy check skips. See core/usersession. package tuic import ( @@ -73,6 +82,7 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo }), tlsConfig: tlsConfig, } + inbound.router = withUserSessions(inbound.router) var udpTimeout time.Duration if options.UDPTimeout != 0 { udpTimeout = time.Duration(options.UDPTimeout) diff --git a/core/protocol/tuic/users.go b/core/protocol/tuic/users.go index 8c4624f4..ad6bece7 100644 --- a/core/protocol/tuic/users.go +++ b/core/protocol/tuic/users.go @@ -1,12 +1,25 @@ package tuic import ( + "github.com/shenaba/2s-ui/core/usersession" + + "github.com/sagernet/sing-box/adapter" "github.com/sagernet/sing-box/option" E "github.com/sagernet/sing/common/exceptions" "github.com/gofrs/uuid/v5" ) +// UpdateUsers swaps the user table of a running inbound and disconnects +// everyone who just left it. Both halves are needed: the table alone only +// decides who may open a new session, while a client that authenticated before +// the change keeps opening streams on the one it already has. +// +// The second half is not something this layer can do to a QUIC session, so when +// a removed user is still connected it reports ErrRestartRequired and the +// caller rebuilds the inbound instead. That disconnects everyone on it once, +// which is why it is reported rather than done unconditionally: an update that +// removes nobody, or removes nobody who is connected, still costs nothing. func (h *Inbound) UpdateUsers(users []option.TUICUser) error { userList := make([]string, 0, len(users)) userUUIDList := make([][16]byte, 0, len(users)) @@ -24,5 +37,22 @@ func (h *Inbound) UpdateUsers(users []option.TUICUser) error { userPasswordList = append(userPasswordList, user.Password) } h.server.UpdateUsers(userList, userUUIDList, userPasswordList) - return nil + return h.sessions().CloseUsers(usersession.KeepSet(userList)).RestartRequired() +} + +// withUserSessions installs the session registry in front of the router. TUIC +// authenticates once per QUIC session and every stream after that rides it, so +// the registry is here to answer one question at removal time: is the user +// being removed actually connected? It cannot do more than that -- sing-quic +// hands out neither a handle on the session nor a stable name for it. See +// core/usersession. +func withUserSessions(router adapter.ConnectionRouterEx) adapter.ConnectionRouterEx { + return usersession.WrapRouterEx(router, usersession.BindOnly) +} + +// sessions reaches the registry withUserSessions installed. The assertion is +// bare on purpose: reordering NewInbound so that it no longer holds would +// otherwise turn every hook here into a silent no-op. +func (h *Inbound) sessions() *usersession.Registry { + return h.router.(*usersession.RouterEx).Registry() } diff --git a/core/protocol/vless/inbound.go b/core/protocol/vless/inbound.go index 9b9b06a4..808715d0 100644 --- a/core/protocol/vless/inbound.go +++ b/core/protocol/vless/inbound.go @@ -5,7 +5,7 @@ // tearing the listener down (see core/inbound_users.go). // // UPGRADING sing-box: re-copy this file from the new tag and re-apply the -// change below. Nothing here will fail to compile if you forget, it will just +// changes below. Nothing here will fail to compile if you forget, it will just // silently keep running the old implementation. // // Local change vs sing-box: the service is keyed by user name @@ -14,6 +14,15 @@ // UpdateUsers does rewrite it, under live sessions, and a position is not. // Deleting a user shifted every later one, which mis-attributed traffic and // could index past the end of the name slice outright (upstream #1231). +// +// Second local change: one line installs the user-session registry +// +// inbound.router = withUserSessions(inbound.router) +// +// which is what lets an already authenticated session be reached when its user +// is removed from the inbound -- swapping the user table alone only decides who +// may start a new one. That line is the whole of it here; everything it reaches +// lives in users.go, which the copy check skips. See core/usersession. package vless import ( @@ -71,6 +80,7 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo if err != nil { return nil, err } + inbound.router = withUserSessions(inbound.router) service := vless.NewService[string](logger, adapter.NewUpstreamContextHandler(inbound.newConnectionEx, inbound.newPacketConnectionEx)) service.UpdateUsers(common.Map(options.Users, func(it option.VLESSUser) string { return it.Name diff --git a/core/protocol/vless/users.go b/core/protocol/vless/users.go index c436ee46..5b905f1a 100644 --- a/core/protocol/vless/users.go +++ b/core/protocol/vless/users.go @@ -1,10 +1,18 @@ package vless import ( + "github.com/shenaba/2s-ui/core/usersession" + + "github.com/sagernet/sing-box/adapter" "github.com/sagernet/sing-box/option" "github.com/sagernet/sing/common" ) +// UpdateUsers swaps the user table of a running inbound and cuts the multiplex +// sessions of everyone who just left it. VLESS authenticates per connection, so +// the table alone already shuts a removed user out of new ones -- except on a +// multiplex session, where one authenticated carrier connection keeps serving +// every stream opened after the change. func (h *Inbound) UpdateUsers(users []option.VLESSUser) error { h.service.UpdateUsers(common.Map(users, func(it option.VLESSUser) string { return it.Name @@ -13,5 +21,27 @@ func (h *Inbound) UpdateUsers(users []option.VLESSUser) error { }), common.Map(users, func(it option.VLESSUser) string { return it.Flow })) + // Unclosable is deliberately not checked: a mux carrier is a net.Conn and + // is cut outright, so a removed user never costs this inbound the rebuild a + // QUIC one does. The only thing that can be counted here is a packet conn + // addressed to the mux destination -- not a carrier, and not worth + // disconnecting everyone else for. + h.sessions().CloseUsers(usersession.KeepSet(common.Map(users, func(it option.VLESSUser) string { + return it.Name + }))) return nil } + +// withUserSessions installs the session registry in front of the router. Only +// the multiplex carrier is tracked: every other connection authenticates on its +// own and is already covered by ConnTracker. See core/usersession. +func withUserSessions(router adapter.ConnectionRouterEx) adapter.ConnectionRouterEx { + return usersession.WrapRouterEx(router, usersession.TrackMuxCarrier) +} + +// sessions reaches the registry withUserSessions installed. The assertion is +// bare on purpose: reordering NewInbound so that it no longer holds would +// otherwise turn every hook here into a silent no-op. +func (h *Inbound) sessions() *usersession.Registry { + return h.router.(*usersession.RouterEx).Registry() +} diff --git a/core/protocol/vmess/inbound.go b/core/protocol/vmess/inbound.go index a9372700..0c3c80c4 100644 --- a/core/protocol/vmess/inbound.go +++ b/core/protocol/vmess/inbound.go @@ -5,7 +5,7 @@ // tearing the listener down (see core/inbound_users.go). // // UPGRADING sing-box: re-copy this file from the new tag and re-apply the -// change below. Nothing here will fail to compile if you forget, it will just +// changes below. Nothing here will fail to compile if you forget, it will just // silently keep running the old implementation. // // Local change vs sing-box: the service is keyed by user name @@ -16,6 +16,15 @@ // could index past the end of the name slice outright (upstream #1231). // The sing-vmess import also needs an explicit `vmess` alias, since this // file's own package is called vmess too. +// +// Second local change: one line installs the user-session registry +// +// inbound.router = withUserSessions(inbound.router) +// +// which is what lets an already authenticated session be reached when its user +// is removed from the inbound -- swapping the user table alone only decides who +// may start a new one. That line is the whole of it here; everything it reaches +// lives in users.go, which the copy check skips. See core/usersession. package vmess import ( @@ -74,6 +83,7 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo if err != nil { return nil, err } + inbound.router = withUserSessions(inbound.router) var serviceOptions []vmess.ServiceOption if timeFunc := ntp.TimeFuncFromContext(ctx); timeFunc != nil { serviceOptions = append(serviceOptions, vmess.ServiceWithTimeFunc(timeFunc)) diff --git a/core/protocol/vmess/users.go b/core/protocol/vmess/users.go index b98d3109..2f32add6 100644 --- a/core/protocol/vmess/users.go +++ b/core/protocol/vmess/users.go @@ -1,16 +1,42 @@ package vmess import ( + "github.com/shenaba/2s-ui/core/usersession" + + "github.com/sagernet/sing-box/adapter" "github.com/sagernet/sing-box/option" "github.com/sagernet/sing/common" ) +// UpdateUsers swaps the user table of a running inbound and cuts the multiplex +// sessions of everyone who just left it -- see the note in vless/users.go. +// Sessions are only cut once the table actually changed, so a rejected update +// leaves the inbound exactly as it was. func (h *Inbound) UpdateUsers(users []option.VMessUser) error { - return h.service.UpdateUsers(common.Map(users, func(it option.VMessUser) string { + err := h.service.UpdateUsers(common.Map(users, func(it option.VMessUser) string { return it.Name }), common.Map(users, func(it option.VMessUser) string { return it.UUID }), common.Map(users, func(it option.VMessUser) int { return it.AlterId })) + if err != nil { + return err + } + // Unclosable is not checked here; see the note in vless/users.go. + h.sessions().CloseUsers(usersession.KeepSet(common.Map(users, func(it option.VMessUser) string { + return it.Name + }))) + return nil +} + +// withUserSessions installs the session registry in front of the router. Only +// the multiplex carrier is tracked: every other connection authenticates on its +// own and is already covered by ConnTracker. See core/usersession. +func withUserSessions(router adapter.ConnectionRouterEx) adapter.ConnectionRouterEx { + return usersession.WrapRouterEx(router, usersession.TrackMuxCarrier) +} + +func (h *Inbound) sessions() *usersession.Registry { + return h.router.(*usersession.RouterEx).Registry() } diff --git a/core/usersession/registry.go b/core/usersession/registry.go new file mode 100644 index 00000000..04c87408 --- /dev/null +++ b/core/usersession/registry.go @@ -0,0 +1,304 @@ +// Package usersession tracks the client sessions of an inbound, so that the +// sessions of a user who was just removed can actually be cut. +// +// Protocols that authenticate once per session (the QUIC ones, anytls, and +// anything carried over sing-mux) keep serving an already authenticated client +// after its user is gone from the inbound: swapping the auth map only affects +// new sessions, and closing the routed connections is not enough because the +// client simply opens another stream on the session it already has. Reaching +// the session itself is what this registry is for. +// +// # Two kinds of session +// +// anytls and the sing-mux carrier arrive as a net.Conn that lives exactly as +// long as the session does. Those are tracked by source address and closed +// outright, and that is the whole story for anytls, vless, vmess and trojan: +// the key is a single TCP connection's address, it cannot move, and Untrack +// runs when the session ends. +// +// The QUIC protocols give this layer no handle on the session -- and, the part +// that decides the design, no usable name for one either: +// +// - sing-quic keeps its session list unexported, and the ctx it hands the +// handler per stream is the Service's own, shared by every session; +// - all three services set quic-go's DisablePathManager, which rewrites a +// connection's remote address the moment a decryptable packet arrives from +// a new one, with no path validation -- so the source address moves under +// one NAT rebind; +// - that address is an ephemeral UDP port, recycled to somebody else once the +// session ends, and nothing tells this layer that it has been; +// - and there is no moment at which a QUIC session can be declared gone. +// quic-go keeps one alive with a 10s PING whether or not a single byte is +// routed, so an idle session and a dead one look identical from here. +// +// So for QUIC this package does not track sessions at all. It records +// something weaker and completely reliable: **which users have been seen on +// this inbound**. That set is bounded by the client count rather than the +// session count, which is what lets it carry no timeout -- and a set with no +// timeout is the only kind that cannot be wrong about a session it cannot see. +// +// Earlier revisions tried to do better and could not. Muting the source address +// is defeated by the second and third points; ageing a per-session entry out +// after ten idle minutes is defeated by the fourth, and silently let a removed +// user keep an idle session. Both looked like they worked because a test client +// that reconnects every second never exercises either. +// +// # What a removal does +// +// CloseUsers cuts every session it holds a transport for, and reports the rest +// as Unclosable. The QUIC inbounds turn that into ErrRestartRequired and the +// caller rebuilds the inbound, which destroys every QUIC session on it, the +// removed user's included. Everyone on that inbound reconnects once. +// +// Being wrong here is one-sided on purpose: a user who connected and then left +// for good is still in the seen set, so removing them costs one restart that +// bought nothing. The other direction -- deciding a session is gone when it is +// not -- is what issue #175 is, so the cost is paid on that side. +// +// This lives at the inbound layer rather than in ConnTracker on purpose. The +// two stay separate: IP limits keep their gate in ConnTracker, because that +// policy has to outlive a core restart, while everything here is per-inbound +// and goes away with the Box. +package usersession + +import ( + "io" + "sync" + + E "github.com/sagernet/sing/common/exceptions" +) + +// entry is one session this layer can actually reach: a transport it can close, +// and the user that authenticated it. +type entry struct { + user string + closer io.Closer +} + +// Registry is what an inbound knows about who is connected to it. One instance +// per inbound; it goes away with the inbound. +// +// Neither map is consulted to decide whether a connection may pass -- nothing +// in this package refuses anything. They decide whose transport to close, and +// whether a removed user was connected at all. +type Registry struct { + access sync.Mutex + // sources holds the sessions with a closer, keyed by the source address of + // the one connection that carries them. Bounded by live sessions: every + // entry is created by Track or BindAndTrack and removed by the Untrack its + // inbound defers. + sources map[string]*entry + // seen holds the users that turned up without a closable session -- the + // QUIC ones. Keyed by user rather than by address because no address here + // is stable, and bounded by the client count rather than by the session + // count, which is what lets it need no expiry. See the package comment. + seen map[string]struct{} +} + +func NewRegistry() *Registry { + return &Registry{ + sources: make(map[string]*entry), + seen: make(map[string]struct{}), + } +} + +// Result is what a removal or a kick managed to do. +type Result struct { + // Cut counts the sessions closed outright. + Cut int + // Unclosable counts the users left connected because this layer has no + // handle on their session. For a QUIC inbound that is the signal to rebuild + // the inbound instead; see ErrRestartRequired. + Unclosable int +} + +// RestartRequired is what an inbound whose sessions it cannot close returns +// from UpdateUsers: nil when the removal was carried out in full, and +// ErrRestartRequired when a removed user is still connected by a session that +// only rebuilding the inbound will end. +// +// Only the QUIC inbounds call this. anytls and the mux protocols hold the +// transport of every session they register -- see the notes on their own +// CloseUsers calls. +func (r Result) RestartRequired() error { + if r.Unclosable > 0 { + return ErrRestartRequired + } + return nil +} + +// Bind records that user is connected. If the connection belongs to a session +// something already tracked (anytls registers its transport in NewConnection +// before any stream is routed), the user is attached to that session so it can +// be closed by name. Otherwise there is no session to attach to and the user +// goes into the seen set -- which is the QUIC path. +func (r *Registry) Bind(user string, source string) { + if user == "" { + return + } + r.access.Lock() + defer r.access.Unlock() + if e, tracked := r.sources[source]; tracked { + e.user = user + return + } + r.seen[user] = struct{}{} +} + +// BindAndTrack records the user and the session transport in one go, for a +// carrier that arrives with both already known. +// +// The two must land under a single lock. A CloseUsers that ran in between -- +// seeing the user but not yet the closer -- would put it in the seen set and +// report it Unclosable, costing a restart that the closer it was about to be +// given would have made unnecessary. +func (r *Registry) BindAndTrack(user string, source string, closer io.Closer) { + if source == "" { + return + } + r.access.Lock() + defer r.access.Unlock() + e := r.load(source) + if user != "" { + e.user = user + } + if closer != nil { + e.closer = closer + } +} + +// Track stores the session transport, for a protocol whose session has a closer +// of its own but does not know the user yet. The Bind that follows attaches it. +func (r *Registry) Track(source string, closer io.Closer) { + if source == "" { + return + } + r.access.Lock() + defer r.access.Unlock() + r.load(source).closer = closer +} + +// Untrack forgets a session that has ended. This is what keeps sources bounded, +// and it is why a tracked session never needs to be aged out: its inbound +// defers this call for exactly as long as the session lives. +func (r *Registry) Untrack(source string) { + if source == "" { + return + } + r.access.Lock() + defer r.access.Unlock() + delete(r.sources, source) +} + +func (r *Registry) load(source string) *entry { + e, loaded := r.sources[source] + if !loaded { + e = &entry{} + r.sources[source] = e + } + return e +} + +// CloseUsers cuts the sessions of every user not in keep, and reports the users +// it could not reach. keep is the user table the inbound has just installed, so +// anything bound to a name outside it belongs to a user who is gone. +func (r *Registry) CloseUsers(keep map[string]struct{}) Result { + r.access.Lock() + var closers []io.Closer + var result Result + for source, e := range r.sources { + if e.user == "" { + continue + } + if _, ok := keep[e.user]; ok { + continue + } + if e.closer == nil { + result.Unclosable++ + continue + } + result.Cut++ + closers = append(closers, e.closer) + delete(r.sources, source) + } + for user := range r.seen { + if _, ok := keep[user]; ok { + continue + } + result.Unclosable++ + // Dropped because the user is off the inbound: either the caller is + // about to rebuild it, which discards this registry anyway, or it is a + // protocol that does not rebuild, where reporting the same departed + // user on every later save would be noise. A user who is re-enabled and + // connects again is recorded again by Bind. + delete(r.seen, user) + } + r.access.Unlock() + + // Outside the lock: closing a tracked session runs the inbound's own close + // handler, which comes back through Untrack. + for _, closer := range closers { + _ = closer.Close() + } + return result +} + +// KickUserSessions disconnects a user who is still enabled -- an operator +// action, not a revocation. Only a session with a closer can be cut; a user who +// is merely known to be connected is reported instead, and it is the caller's +// business whether disconnecting one user is worth restarting the inbound +// everyone else is on. +func (r *Registry) KickUserSessions(user string) Result { + if user == "" { + return Result{} + } + + r.access.Lock() + var closers []io.Closer + var result Result + for source, e := range r.sources { + if e.user != user { + continue + } + if e.closer == nil { + result.Unclosable++ + continue + } + result.Cut++ + closers = append(closers, e.closer) + delete(r.sources, source) + } + // Left in the set: this user is still on the inbound and may still be + // connected, so a later removal has to report them again. + if _, ok := r.seen[user]; ok { + result.Unclosable++ + } + r.access.Unlock() + + for _, closer := range closers { + _ = closer.Close() + } + return result +} + +// KeepSet turns the user-name list an inbound just installed into the set +// CloseUsers takes. The names an in-place update is given are exactly the +// clients still enabled on that inbound, so nothing else has to be looked up. +func KeepSet(names []string) map[string]struct{} { + keep := make(map[string]struct{}, len(names)) + for _, name := range names { + keep[name] = struct{}{} + } + return keep +} + +// ErrRestartRequired reports that a user who was just removed is still +// connected by a session this layer cannot close, so the only way to disconnect +// them is to tear the inbound down and build it again. +// +// The caller already has that path -- it is the same fallback taken by +// protocols with no in-place user update at all -- so returning this error is +// all an inbound has to do. It is not a failure: the user table was swapped +// successfully, and the caller tells the two apart so that this one is not +// logged as one. +var ErrRestartRequired = E.New("removed user is still connected by a session that cannot be cut") diff --git a/core/usersession/registry_test.go b/core/usersession/registry_test.go new file mode 100644 index 00000000..cd706b1d --- /dev/null +++ b/core/usersession/registry_test.go @@ -0,0 +1,326 @@ +package usersession + +import ( + "errors" + "strconv" + "testing" +) + +type fakeCloser struct { + closed bool +} + +func (c *fakeCloser) Close() error { + c.closed = true + return nil +} + +func keep(users ...string) map[string]struct{} { + set := make(map[string]struct{}, len(users)) + for _, user := range users { + set[user] = struct{}{} + } + return set +} + +// What the QUIC inbounds turn a CloseUsers result into. Asking for a restart +// when nothing was left connected would disconnect a whole inbound on every +// ordinary save; not asking when something was leaves the removed user online. +func TestResultRestartRequired(t *testing.T) { + for _, c := range []struct { + result Result + want error + }{ + {Result{}, nil}, + {Result{Cut: 3}, nil}, + {Result{Unclosable: 1}, ErrRestartRequired}, + {Result{Cut: 2, Unclosable: 1}, ErrRestartRequired}, + } { + if got := c.result.RestartRequired(); !errors.Is(got, c.want) { + t.Errorf("%+v.RestartRequired() = %v, want %v", c.result, got, c.want) + } + } +} + +func TestCloseUsersClosesTrackedSession(t *testing.T) { + r := NewRegistry() + conn := &fakeCloser{} + r.Track("1.2.3.4:1000", conn) + r.Bind("gone", "1.2.3.4:1000") + + got := r.CloseUsers(keep("stays")) + if got != (Result{Cut: 1}) { + t.Fatalf("CloseUsers = %+v, want {Cut:1}", got) + } + if !conn.closed { + t.Error("a tracked session must be closed") + } +} + +// The QUIC shape: nothing was tracked, so there is no session to cut. Saying so +// is the whole contract -- it is what makes the inbound ask to be rebuilt. +func TestCloseUsersReportsUntrackedUser(t *testing.T) { + r := NewRegistry() + r.Bind("gone", "1.2.3.4:1000") + + got := r.CloseUsers(keep("stays")) + if got != (Result{Unclosable: 1}) { + t.Fatalf("CloseUsers = %+v, want {Unclosable:1}", got) + } +} + +// THE regression this package exists for, and the one an earlier revision got +// wrong twice. A QUIC session that has routed nothing for a long time is not a +// session that ended: quic-go keeps it alive with a 10s PING, and the streams +// it opens later never re-check the user table. Nothing may therefore expire a +// user out of the seen set -- only their removal takes them out of it. +// +// There is no clock to advance here, which is the point: the fix was to delete +// the notion of an idle session rather than to tune it. What stands in for +// elapsed time is unbounded unrelated activity. +func TestIdleUserIsNeverForgotten(t *testing.T) { + r := NewRegistry() + r.Bind("idle", "1.2.3.4:1000") + + // Lots of other traffic and lots of ordinary saves, none of which concern + // "idle" -- who sends nothing at all for the whole stretch. + for i := 0; i < 500; i++ { + r.Bind("busy", "5.6.7.8:"+strconv.Itoa(2000+i)) + r.CloseUsers(keep("idle", "busy")) + } + + got := r.CloseUsers(keep("busy")) + if got != (Result{Unclosable: 1}) { + t.Fatalf("CloseUsers = %+v, want {Unclosable:1} -- an idle user was forgotten", got) + } +} + +// The seen set is keyed by user, not by address, and that is what makes it safe +// to keep forever: it is bounded by the client count, not by how many sessions +// or source ports a client burns through. A QUIC client whose address moves +// (NAT rebind, DisablePathManager) is also still the same one entry. +func TestSeenSetIsKeyedByUser(t *testing.T) { + r := NewRegistry() + for i := 0; i < 1000; i++ { + r.Bind("roamer", "1.2.3.4:"+strconv.Itoa(1000+i)) + } + + r.access.Lock() + size := len(r.seen) + sources := len(r.sources) + r.access.Unlock() + if size != 1 { + t.Errorf("len(seen) = %d after 1000 addresses, want 1", size) + } + if sources != 0 { + t.Errorf("len(sources) = %d, want 0 -- an untracked user must not allocate per address", sources) + } + + got := r.CloseUsers(keep()) + if got != (Result{Unclosable: 1}) { + t.Errorf("CloseUsers = %+v, want {Unclosable:1} -- one user is one report", got) + } +} + +// Re-enabling a user has to work: their next connection puts them back. +func TestRemovedUserIsRecordedAgainOnReconnect(t *testing.T) { + r := NewRegistry() + r.Bind("flip", "1.2.3.4:1000") + if got := r.CloseUsers(keep()); got != (Result{Unclosable: 1}) { + t.Fatalf("precondition: CloseUsers = %+v, want {Unclosable:1}", got) + } + // Reported once and dropped, so an unrelated later save says nothing. + if got := r.CloseUsers(keep()); got != (Result{}) { + t.Fatalf("CloseUsers = %+v on the second pass, want an empty result", got) + } + + r.Bind("flip", "1.2.3.4:2000") + if got := r.CloseUsers(keep()); got != (Result{Unclosable: 1}) { + t.Errorf("CloseUsers = %+v after reconnect, want {Unclosable:1}", got) + } +} + +func TestCloseUsersLeavesUnauthenticatedSourceAlone(t *testing.T) { + r := NewRegistry() + // Bound with no user name: the connection reached the handler before + // authentication produced one. It belongs to nobody, so a removal has + // nothing to say about it. + r.Bind("", "1.2.3.4:1000") + + got := r.CloseUsers(keep("stays")) + if got != (Result{}) { + t.Fatalf("CloseUsers = %+v, want an empty result", got) + } +} + +// A user still on the inbound is not touched, however many sessions they have. +func TestCloseUsersKeepsEnabledUsers(t *testing.T) { + r := NewRegistry() + first := &fakeCloser{} + second := &fakeCloser{} + r.BindAndTrack("stays", "1.2.3.4:1000", first) + r.BindAndTrack("stays", "1.2.3.4:1001", second) + r.Bind("stays", "5.6.7.8:2000") + + got := r.CloseUsers(keep("stays")) + if got != (Result{}) { + t.Fatalf("CloseUsers = %+v, want an empty result", got) + } + if first.closed || second.closed { + t.Error("a user who is still enabled must keep their sessions") + } +} + +func TestKickClosesTrackedSession(t *testing.T) { + r := NewRegistry() + conn := &fakeCloser{} + r.Track("1.2.3.4:1000", conn) + r.Bind("noisy", "1.2.3.4:1000") + + got := r.KickUserSessions("noisy") + if got != (Result{Cut: 1}) { + t.Fatalf("KickUserSessions = %+v, want {Cut:1}", got) + } + if !conn.closed { + t.Error("a tracked session must be closed on kick") + } +} + +// A kick cannot reach a QUIC session either. It says so rather than pretending, +// and leaves what to do about it to the caller. +func TestKickReportsUntrackedUser(t *testing.T) { + r := NewRegistry() + r.Bind("noisy", "1.2.3.4:1000") + + got := r.KickUserSessions("noisy") + if got != (Result{Unclosable: 1}) { + t.Fatalf("KickUserSessions = %+v, want {Unclosable:1}", got) + } + // Unlike a removal, a kick leaves the user in the set: they are still on + // the inbound, so a removal later still has to report them. + if got := r.CloseUsers(keep()); got != (Result{Unclosable: 1}) { + t.Errorf("CloseUsers = %+v after a kick, want {Unclosable:1} -- the kick dropped a user who is still enabled", got) + } +} + +// The other shape a kick can fail to reach: an entry that was tracked but +// carries no closer (a packet conn addressed to the mux destination). It has to +// be reported for the same reason the seen set is. +func TestKickReportsATrackedEntryWithNoCloser(t *testing.T) { + r := NewRegistry() + r.BindAndTrack("live", "1.2.3.4:1000", nil) + + got := r.KickUserSessions("live") + if got != (Result{Unclosable: 1}) { + t.Fatalf("KickUserSessions = %+v, want {Unclosable:1}", got) + } +} + +func TestKickIgnoresOtherUsers(t *testing.T) { + r := NewRegistry() + conn := &fakeCloser{} + r.BindAndTrack("bystander", "1.2.3.4:1000", conn) + r.Bind("other", "5.6.7.8:2000") + + got := r.KickUserSessions("noisy") + if got != (Result{}) { + t.Fatalf("KickUserSessions = %+v, want an empty result", got) + } + if conn.closed { + t.Error("a kick must not touch another user's session") + } +} + +func TestUntrackForgetsSession(t *testing.T) { + r := NewRegistry() + conn := &fakeCloser{} + r.Track("1.2.3.4:1000", conn) + r.Bind("gone", "1.2.3.4:1000") + r.Untrack("1.2.3.4:1000") + + got := r.CloseUsers(keep()) + if got != (Result{}) { + t.Fatalf("CloseUsers = %+v, want an empty result -- the session was already gone", got) + } + if conn.closed { + t.Error("an untracked session must not be closed again") + } +} + +// A tracked session is alive however quiet it has been -- it carries a single +// long-lived stream, an ssh session or a download, and opens nothing new. It +// must stay closable, which is why nothing ages sources out: Untrack is the +// only thing that removes an entry, and it runs when the session really ends. +func TestQuietTrackedSessionStaysClosable(t *testing.T) { + r := NewRegistry() + conn := &fakeCloser{} + r.BindAndTrack("quiet", "1.2.3.4:1000", conn) + + // Plenty of unrelated activity while "quiet" routes nothing. + for i := 0; i < 500; i++ { + r.Bind("busy", "5.6.7.8:"+strconv.Itoa(2000+i)) + r.CloseUsers(keep("quiet", "busy")) + } + + got := r.CloseUsers(keep("busy")) + if got != (Result{Cut: 1}) { + t.Fatalf("CloseUsers = %+v, want {Cut:1} -- the quiet session was lost", got) + } + if !conn.closed { + t.Error("a quiet tracked session must still be closable") + } +} + +// This pins the outcome -- a carrier registered this way is cut, not reported. +// The atomicity it exists for cannot be asserted here: it comes from doing both +// writes under one lock, and a sequential Bind-then-Track would pass this test +// just the same. What that split would lose is the interleaving where a +// CloseUsers lands between the two, sees a user with no session yet, and puts +// them in the seen set -- costing a mux inbound a restart it never needed. +func TestBindAndTrackCutsRatherThanReports(t *testing.T) { + r := NewRegistry() + conn := &fakeCloser{} + r.BindAndTrack("live", "1.2.3.4:1000", conn) + + got := r.CloseUsers(keep()) + if got != (Result{Cut: 1}) { + t.Fatalf("CloseUsers = %+v, want {Cut:1}", got) + } + if !conn.closed { + t.Error("the carrier must be closed") + } +} + +// A packet conn cannot be a mux carrier, so the router passes a nil closer. It +// must stay nil: a nil net.Conn placed in an io.Closer is not a nil interface, +// and would register as a closer that panics when the session is cut. +func TestBindAndTrackWithNilCloserReportsUnclosable(t *testing.T) { + r := NewRegistry() + r.BindAndTrack("live", "1.2.3.4:1000", nil) + + got := r.CloseUsers(keep()) + if got != (Result{Unclosable: 1}) { + t.Fatalf("CloseUsers = %+v, want {Unclosable:1}", got) + } +} + +// Sessions are counted one by one because each is a separate transport to +// close; users without one are counted once, because one restart ends all of +// theirs at the same time. +func TestCloseUsersCountsSessionsAndUsers(t *testing.T) { + r := NewRegistry() + first := &fakeCloser{} + second := &fakeCloser{} + r.BindAndTrack("gone", "1.2.3.4:1000", first) + r.BindAndTrack("gone", "1.2.3.4:1001", second) + r.Bind("gone", "1.2.3.4:1002") + r.Bind("gone", "1.2.3.4:1003") + + got := r.CloseUsers(keep()) + if got != (Result{Cut: 2, Unclosable: 1}) { + t.Fatalf("CloseUsers = %+v, want {Cut:2 Unclosable:1}", got) + } + if !first.closed || !second.closed { + t.Error("both tracked sessions must be cut") + } +} diff --git a/core/usersession/router.go b/core/usersession/router.go new file mode 100644 index 00000000..6ea98331 --- /dev/null +++ b/core/usersession/router.go @@ -0,0 +1,182 @@ +package usersession + +import ( + "context" + "io" + "net" + + "github.com/sagernet/sing-box/adapter" + N "github.com/sagernet/sing/common/network" + + singmux "github.com/sagernet/sing-mux" +) + +// The registry is reached by wrapping an inbound's router rather than by giving +// each inbound a field and a hook in every handler. Both see the same +// connections at the same point, but the copies under core/protocol/ are diffed +// against sing-box line for line by scripts/check-protocol-copies.sh, and this +// way each one carries a single added line instead of a block in every handler. +// +// Nothing here refuses anything: a wrapped router records what it sees and +// passes it on. Refusing by source address is what the package used to do and +// could not be made correct -- see the package comment. +// +// Every copy assigns metadata.User before calling the router, so the user a +// connection authenticated as is already known here. + +// Mode is what a wrapped router does with the connections it sees. Which one an +// inbound wants follows from how its protocol reuses an authenticated +// connection. +type Mode int + +const ( + // BindOnly records which user is connected from where, and nothing else. + // + // The QUIC inbounds take this because they have no session handle to give: + // what the registry does for them is answer, at removal time, whether the + // user being removed is connected at all -- which is what decides between + // a free in-place update and rebuilding the inbound. + // + // anytls takes it too, for the opposite reason: it does have a closer, but + // it holds it in NewConnection, which the router never sees, so it + // registers the session there instead. + BindOnly Mode = iota + // TrackMuxCarrier ignores everything but a sing-mux carrier connection. + // These protocols authenticate per connection, so a removed user is already + // locked out of new ones and ConnTracker closes the routed ones -- except + // on a multiplex session, where the carrier authenticated once and every + // stream after that rides it. The carrier is a net.Conn we can close, so it + // is tracked and cut. + TrackMuxCarrier +) + +type hooks struct { + registry *Registry + mode Mode +} + +// enter records one connection on its way to the real router, and returns the +// cleanup to run once the router is done with it, or nil if there is none. +func (h hooks) enter(conn io.Closer, metadata adapter.InboundContext) func() { + switch h.mode { + case TrackMuxCarrier: + if metadata.Destination != singmux.Destination { + return nil + } + source := metadata.Source.String() + // Only a stream-oriented carrier can be closed; a packet conn reaching + // here would not be a mux carrier anyway. Assigned through a nil-able + // io.Closer rather than passed directly: a nil net.Conn put into an + // interface is not a nil interface, and would register as a closer that + // panics when the session is cut. + var carrier io.Closer + if streamConn, ok := conn.(net.Conn); ok { + carrier = streamConn + } + // One call, one lock: see BindAndTrack for why these cannot be split. + h.registry.BindAndTrack(metadata.User, source, carrier) + // The router blocks for as long as the multiplex session lives, so + // untracking when it returns is not early. + return func() { h.registry.Untrack(source) } + default: + h.registry.Bind(metadata.User, metadata.Source.String()) + return nil + } +} + +// RouterEx wraps an inbound whose router field is an adapter.ConnectionRouterEx. +type RouterEx struct { + adapter.ConnectionRouterEx + hooks +} + +// WrapRouterEx returns router with the session hooks in front of it. +func WrapRouterEx(router adapter.ConnectionRouterEx, mode Mode) *RouterEx { + return &RouterEx{ + ConnectionRouterEx: router, + hooks: hooks{registry: NewRegistry(), mode: mode}, + } +} + +// Registry is how the owning inbound reaches the registry it just installed. +func (r *RouterEx) Registry() *Registry { + return r.registry +} + +func (r *RouterEx) RouteConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { + if done := r.enter(conn, metadata); done != nil { + defer done() + } + r.ConnectionRouterEx.RouteConnectionEx(ctx, conn, metadata, onClose) +} + +func (r *RouterEx) RoutePacketConnectionEx(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { + if done := r.enter(conn, metadata); done != nil { + defer done() + } + r.ConnectionRouterEx.RoutePacketConnectionEx(ctx, conn, metadata, onClose) +} + +// The deprecated pair is overridden too, so that a transport still routing +// through it cannot slip a session past the hooks unrecorded. +func (r *RouterEx) RouteConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error { + if done := r.enter(conn, metadata); done != nil { + defer done() + } + return r.ConnectionRouterEx.RouteConnection(ctx, conn, metadata) +} + +func (r *RouterEx) RoutePacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext) error { + if done := r.enter(conn, metadata); done != nil { + defer done() + } + return r.ConnectionRouterEx.RoutePacketConnection(ctx, conn, metadata) +} + +// Router wraps an inbound whose router field is the full adapter.Router, which +// hysteria and hysteria2 hold. Everything it does not override is forwarded by +// the embedded interface. +type Router struct { + adapter.Router + hooks +} + +// WrapRouter returns router with the session hooks in front of it. +func WrapRouter(router adapter.Router, mode Mode) *Router { + return &Router{ + Router: router, + hooks: hooks{registry: NewRegistry(), mode: mode}, + } +} + +func (r *Router) Registry() *Registry { + return r.registry +} + +func (r *Router) RouteConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { + if done := r.enter(conn, metadata); done != nil { + defer done() + } + r.Router.RouteConnectionEx(ctx, conn, metadata, onClose) +} + +func (r *Router) RoutePacketConnectionEx(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { + if done := r.enter(conn, metadata); done != nil { + defer done() + } + r.Router.RoutePacketConnectionEx(ctx, conn, metadata, onClose) +} + +func (r *Router) RouteConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error { + if done := r.enter(conn, metadata); done != nil { + defer done() + } + return r.Router.RouteConnection(ctx, conn, metadata) +} + +func (r *Router) RoutePacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext) error { + if done := r.enter(conn, metadata); done != nil { + defer done() + } + return r.Router.RoutePacketConnection(ctx, conn, metadata) +} diff --git a/core/usersession/router_test.go b/core/usersession/router_test.go new file mode 100644 index 00000000..9ae33459 --- /dev/null +++ b/core/usersession/router_test.go @@ -0,0 +1,281 @@ +package usersession + +import ( + "context" + "io" + "net" + "os" + "testing" + "time" + + "github.com/sagernet/sing-box/adapter" + "github.com/sagernet/sing/common/buf" + M "github.com/sagernet/sing/common/metadata" + N "github.com/sagernet/sing/common/network" + + singmux "github.com/sagernet/sing-mux" +) + +// fakeRouter is the next hop. It records what reached it and, while it is +// "routing", lets a test look at the registry -- which is how the carrier's +// lifetime is asserted: a mux carrier must be tracked for exactly as long as +// the router call lasts. +type fakeRouter struct { + calls int + last adapter.InboundContext + whileIn func() +} + +func (r *fakeRouter) enter(metadata adapter.InboundContext) { + r.calls++ + r.last = metadata + if r.whileIn != nil { + r.whileIn() + } +} + +func (r *fakeRouter) RouteConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error { + r.enter(metadata) + return nil +} + +func (r *fakeRouter) RoutePacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext) error { + r.enter(metadata) + return nil +} + +func (r *fakeRouter) RouteConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { + r.enter(metadata) +} + +func (r *fakeRouter) RoutePacketConnectionEx(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { + r.enter(metadata) +} + +// fakePacketConn is the smallest thing that satisfies N.PacketConn. Only Close +// is ever reached; the rest exist to satisfy the interface. +type fakePacketConn struct { + closed bool +} + +func (c *fakePacketConn) ReadPacket(buffer *buf.Buffer) (M.Socksaddr, error) { + return M.Socksaddr{}, io.EOF +} +func (c *fakePacketConn) WritePacket(buffer *buf.Buffer, destination M.Socksaddr) error { return nil } +func (c *fakePacketConn) Close() error { c.closed = true; return nil } +func (c *fakePacketConn) LocalAddr() net.Addr { return M.Socksaddr{} } +func (c *fakePacketConn) SetDeadline(t time.Time) error { return nil } +func (c *fakePacketConn) SetReadDeadline(t time.Time) error { return nil } +func (c *fakePacketConn) SetWriteDeadline(t time.Time) error { return nil } + +func metadataFor(user string, source string, destination M.Socksaddr) adapter.InboundContext { + return adapter.InboundContext{ + User: user, + Source: M.ParseSocksaddr(source), + Destination: destination, + } +} + +var plainDestination = M.ParseSocksaddr("example.com:443") + +// closed reports whether one end of a pipe was closed, by reading from the +// other end. The deadline is what makes "still open" answerable at all: a pipe +// nobody closed would otherwise block this read forever. +func closed(t *testing.T, peer net.Conn) bool { + t.Helper() + // net.Pipe refuses a deadline once either end is closed, so this failing is + // itself the answer. + if err := peer.SetReadDeadline(time.Now().Add(200 * time.Millisecond)); err != nil { + return true + } + buf := make([]byte, 1) + _, err := peer.Read(buf) + if err == nil || os.IsTimeout(err) { + return false + } + return true +} + +// Nothing here refuses a connection, in any mode: refusing by source address is +// what this package used to do and could not be made correct. Every mode is +// asserted against that below -- a gate creeping back in is the regression this +// file exists to catch. +func TestNoModeEverRefuses(t *testing.T) { + for _, mode := range []struct { + name string + mode Mode + }{ + {"BindOnly", BindOnly}, + {"TrackMuxCarrier", TrackMuxCarrier}, + } { + t.Run(mode.name, func(t *testing.T) { + next := &fakeRouter{} + router := WrapRouterEx(next, mode.mode) + // Learn a user, then remove them. Whatever the registry made of + // that, the next connection from the same address still goes + // through: for TrackMuxCarrier because trojan's fallback shares + // this router, and for BindOnly because the address may by now + // belong to somebody else entirely. + router.Registry().Bind("gone", "1.2.3.4:1000") + router.Registry().CloseUsers(KeepSet(nil)) + + conn, peer := net.Pipe() + defer conn.Close() + defer peer.Close() + var reported error + router.RouteConnectionEx(context.Background(), conn, metadataFor("gone", "1.2.3.4:1000", plainDestination), func(err error) { reported = err }) + + if next.calls != 1 { + t.Errorf("router calls = %d, want 1 -- the connection was refused", next.calls) + } + if reported != nil { + t.Errorf("close handler got %v, want nothing", reported) + } + if closed(t, peer) { + t.Error("the connection was closed") + } + }) + } +} + +func TestBindOnlyRecordsTheUser(t *testing.T) { + next := &fakeRouter{} + router := WrapRouterEx(next, BindOnly) + + conn, peer := net.Pipe() + defer conn.Close() + defer peer.Close() + router.RouteConnectionEx(context.Background(), conn, metadataFor("live", "1.2.3.4:1000", plainDestination), nil) + + if next.calls != 1 { + t.Fatalf("router calls = %d, want 1", next.calls) + } + // Bound, so a removal knows this user is connected -- but with no closer, + // because the session this connection belongs to is either a QUIC one + // (unreachable) or an anytls one registered elsewhere. + got := router.Registry().CloseUsers(KeepSet(nil)) + if got != (Result{Unclosable: 1}) { + t.Errorf("CloseUsers = %+v, want {Unclosable:1}", got) + } + if closed(t, peer) { + t.Error("BindOnly must not close the connection it saw") + } +} + +// The deprecated pair is overridden so a transport still routing through it +// cannot slip a session past unrecorded. +func TestDeprecatedRoutePathIsRecorded(t *testing.T) { + next := &fakeRouter{} + router := WrapRouterEx(next, BindOnly) + + conn, peer := net.Pipe() + defer conn.Close() + defer peer.Close() + err := router.RouteConnection(context.Background(), conn, metadataFor("live", "1.2.3.4:1000", plainDestination)) + if err != nil { + t.Fatalf("RouteConnection returned %v", err) + } + if next.calls != 1 { + t.Fatalf("router calls = %d, want 1", next.calls) + } + got := router.Registry().CloseUsers(KeepSet(nil)) + if got != (Result{Unclosable: 1}) { + t.Errorf("CloseUsers = %+v, want {Unclosable:1} -- the deprecated path did not record the user", got) + } +} + +func TestMuxCarrierIsTrackedForTheRoutersLifetime(t *testing.T) { + var trackedDuring bool + next := &fakeRouter{} + router := WrapRouterEx(next, TrackMuxCarrier) + next.whileIn = func() { + // The router blocks for as long as the multiplex session lives, so the + // carrier has to be closable right here -- that is the whole point. + trackedDuring = router.Registry().KickUserSessions("live") == (Result{Cut: 1}) + } + + conn, peer := net.Pipe() + defer peer.Close() + router.RouteConnectionEx(context.Background(), conn, metadataFor("live", "1.2.3.4:1000", singmux.Destination), nil) + + if !trackedDuring { + t.Error("the mux carrier was not closable while the router had it") + } + if !closed(t, peer) { + t.Error("kicking the user must have closed the carrier") + } +} + +// The carrier is forgotten once its session ends. Left behind, the entry +// outlives the session it names, and the next removal cuts a connection that +// is not there any more -- or, on an inbound that reports instead, asks for a +// restart on behalf of a client that already left. +// +// Kicking first (as the test above does) deletes the entry by itself, so this +// has to be its own case to mean anything. +func TestMuxCarrierIsUntrackedWhenTheSessionEnds(t *testing.T) { + next := &fakeRouter{} + router := WrapRouterEx(next, TrackMuxCarrier) + + conn, peer := net.Pipe() + defer conn.Close() + defer peer.Close() + // fakeRouter returns immediately, which stands in for the session ending. + router.RouteConnectionEx(context.Background(), conn, metadataFor("live", "1.2.3.4:1000", singmux.Destination), nil) + + got := router.Registry().CloseUsers(KeepSet(nil)) + if got != (Result{}) { + t.Errorf("CloseUsers = %+v, want an empty result -- the carrier outlived its session", got) + } +} + +func TestMuxModeIgnoresPlainConnections(t *testing.T) { + next := &fakeRouter{} + router := WrapRouterEx(next, TrackMuxCarrier) + var registeredDuring Result + next.whileIn = func() { + // Asked while the router still has the connection. Asking afterwards + // proves nothing: the deferred Untrack empties the registry either way, + // so the assertion would hold even if every connection were registered. + registeredDuring = router.Registry().CloseUsers(KeepSet(nil)) + } + + conn, peer := net.Pipe() + defer conn.Close() + defer peer.Close() + router.RouteConnectionEx(context.Background(), conn, metadataFor("live", "1.2.3.4:1000", plainDestination), nil) + + if next.calls != 1 { + t.Fatalf("router calls = %d, want 1", next.calls) + } + // A plain connection authenticates on its own and ConnTracker closes it, so + // it has no business in the registry. + if registeredDuring != (Result{}) { + t.Errorf("CloseUsers = %+v mid-route, want an empty result -- a plain connection was registered", registeredDuring) + } + if closed(t, peer) { + t.Error("a plain connection must not be closed by this layer") + } +} + +// A packet conn cannot be a mux carrier. If one is registered as though it +// were, a removal closes it -- so the carrier slot has to stay empty, which is +// also what keeps a nil net.Conn out of a non-nil io.Closer. +func TestMuxModeDoesNotTakeAPacketConnAsCarrier(t *testing.T) { + next := &fakeRouter{} + router := WrapRouterEx(next, TrackMuxCarrier) + conn := &fakePacketConn{} + var registeredDuring Result + next.whileIn = func() { + registeredDuring = router.Registry().CloseUsers(KeepSet(nil)) + } + + router.RoutePacketConnectionEx(context.Background(), conn, metadataFor("live", "1.2.3.4:1000", singmux.Destination), nil) + + if registeredDuring != (Result{Unclosable: 1}) { + t.Errorf("CloseUsers = %+v mid-route, want {Unclosable:1}", registeredDuring) + } + if conn.closed { + t.Error("a packet conn was closed as though it were a mux carrier") + } +} diff --git a/core/usersession/stress_test.go b/core/usersession/stress_test.go new file mode 100644 index 00000000..60cda6e3 --- /dev/null +++ b/core/usersession/stress_test.go @@ -0,0 +1,58 @@ +package usersession + +import ( + "strconv" + "sync" + "testing" +) + +// Every exported method driven at once from many goroutines. The point is the +// race detector, not the assertions: CloseUsers closes outside the lock and +// Untrack comes back in on the closer's own goroutine, so the two have to be +// safe against each other. Run with -race to mean anything. +func TestRegistryUnderConcurrency(t *testing.T) { + r := NewRegistry() + const workers = 8 + const rounds = 200 + + var wg sync.WaitGroup + for w := 0; w < workers; w++ { + wg.Add(1) + go func(w int) { + defer wg.Done() + user := "user" + strconv.Itoa(w%3) + for i := 0; i < rounds; i++ { + source := "10.0.0." + strconv.Itoa(w) + ":" + strconv.Itoa(1000+i%7) + switch i % 5 { + case 0: + r.Bind(user, source) + case 1: + r.BindAndTrack(user, source, &fakeCloser{}) + case 2: + r.Track(source, &fakeCloser{}) + r.Bind(user, source) + case 3: + r.CloseUsers(keep("user0")) + case 4: + r.KickUserSessions(user) + } + r.Untrack(source) + } + }(w) + } + wg.Wait() + + // Every source was untracked by its own goroutine, so nothing may be left. + r.access.Lock() + remaining := len(r.sources) + seen := len(r.seen) + r.access.Unlock() + if remaining != 0 { + t.Errorf("len(sources) = %d after every source was untracked, want 0", remaining) + } + // seen has no expiry, so what bounds it has to be the user count: three + // users across 8 goroutines and 1600 addresses. + if seen > 3 { + t.Errorf("len(seen) = %d, want at most 3 -- it is growing per address, not per user", seen) + } +} diff --git a/go.mod b/go.mod index 476d2f45..27317995 100644 --- a/go.mod +++ b/go.mod @@ -17,6 +17,7 @@ require ( github.com/robfig/cron/v3 v3.0.1 github.com/sagernet/sing v0.9.4 github.com/sagernet/sing-box v1.14.1 + github.com/sagernet/sing-mux v0.3.6 github.com/sagernet/sing-quic v0.7.0 github.com/sagernet/sing-tun v0.9.3 github.com/sagernet/sing-vmess v0.2.8 @@ -169,7 +170,6 @@ require ( github.com/sagernet/nftables v0.3.0-mod.4 // indirect github.com/sagernet/quic-go v0.61.0-sing-box-mod.7 // indirect github.com/sagernet/sing-cloudflared v0.1.3-0.20260706062323-d9787e794aa3 // indirect - github.com/sagernet/sing-mux v0.3.6 // indirect github.com/sagernet/sing-openconnect v0.0.0-20260810065514-53aa8058f8df // indirect github.com/sagernet/sing-openvpn v0.0.0-20260729104525-103eb5fe5eb6 // indirect github.com/sagernet/sing-shadowsocks v0.2.8 // indirect diff --git a/scripts/check-protocol-copies.sh b/scripts/check-protocol-copies.sh index f204fe66..fa922a1f 100644 --- a/scripts/check-protocol-copies.sh +++ b/scripts/check-protocol-copies.sh @@ -32,28 +32,39 @@ echo "comparing core/protocol/ against sing-box $VERSION" LOCAL_ONLY="users.go" # Lines each copy is expected to differ by, beyond the shared header comment, -# keyed by "/". +# keyed by "/". These are not a tolerance: each is the exact +# size of changes we made on purpose, so a bump that alters the file anywhere +# else still shows up. Two changes make them up. # -# The six user-carrying protocols key their service by user name rather than by -# list position (Service[string], not Service[int]) -- see the header of any of -# those files. That patch is what these counts are: they are not a tolerance, -# they are the exact size of a change we made on purpose, so a bump that alters -# the file elsewhere still shows up. anytls has no user table and stays verbatim. +# 1. The six user-carrying protocols key their service by user name rather than +# by list position (Service[string], not Service[int]) -- see the header of +# any of those files. vmess also carries 2 lines of import alias, since its +# own package is `vmess`. anytls has no user table and is untouched by this. # -# vmess also carries 2 lines of import alias, since its own package is `vmess`. +# 2. Every copy installs the user-session registry by wrapping its router, which +# is one added line, identical in all seven: # -# Re-copying after a sing-box bump will change these. Re-apply the patch, then -# put the new counts here -- and read the diff first rather than just pasting -# the number the script printed, which is how a real upstream change gets -# rubber-stamped into the expected total. +# inbound.router = withUserSessions(inbound.router) +# +# Everything that line reaches lives in users.go, which is exempt below. The +# hook is a wrapper rather than a field plus a block in each handler exactly +# so that these counts stay this small -- see core/usersession. anytls costs +# 3 instead of 1: its session is held in NewConnection, which the router +# never sees, so that one call is redirected through users.go as well. +# +# Re-copying after a sing-box bump will change these. Re-apply both changes, +# then put the new counts here -- and read the diff first rather than just +# pasting the number the script printed, which is how a real upstream change +# gets rubber-stamped into the expected total. expect_diff() { case "$1" in - hysteria/inbound.go) echo 28 ;; - hysteria2/inbound.go) echo 28 ;; - trojan/inbound.go) echo 25 ;; - tuic/inbound.go) echo 26 ;; - vless/inbound.go) echo 29 ;; - vmess/inbound.go) echo 27 ;; + anytls/inbound.go) echo 3 ;; + hysteria/inbound.go) echo 29 ;; + hysteria2/inbound.go) echo 29 ;; + trojan/inbound.go) echo 26 ;; + tuic/inbound.go) echo 27 ;; + vless/inbound.go) echo 30 ;; + vmess/inbound.go) echo 28 ;; *) echo 0 ;; esac } diff --git a/service/inbounds.go b/service/inbounds.go index 48ee9674..6bc94ef6 100644 --- a/service/inbounds.go +++ b/service/inbounds.go @@ -2,11 +2,13 @@ package service import ( "encoding/json" + "errors" "fmt" "os" "strconv" "strings" + "github.com/shenaba/2s-ui/core/usersession" "github.com/shenaba/2s-ui/database" "github.com/shenaba/2s-ui/database/model" "github.com/shenaba/2s-ui/logger" @@ -463,13 +465,22 @@ func (s *InboundService) UpdateInboundsUsers(tx *gorm.DB, ids []uint) error { return err } - // An in-place update that errors leaves the inbound running with its old + // Two different things arrive here as an error, and the restart below + // answers both. A real failure leaves the inbound running with its old // user table, so a removed user would stay connected and keep - // authenticating. Fall through to the restart rather than returning: - // dropping every connection on this inbound is the safe failure. + // authenticating -- dropping every connection on the inbound is the + // safe failure. ErrRestartRequired is the opposite: the table was + // swapped fine, and the inbound is asking to be rebuilt because a user + // we just removed holds a session it cannot close (a QUIC one; see + // core/usersession). Only the first is something going wrong, so only + // the first is logged as such. handled, err := corePtr.UpdateInboundUsers(inboundConfig) if err != nil { - logger.Warning("in-place user update failed for inbound ", inbound.Tag, ", restarting it: ", err) + if errors.Is(err, usersession.ErrRestartRequired) { + logger.Info("restarting inbound ", inbound.Tag, ": ", err) + } else { + logger.Warning("in-place user update failed for inbound ", inbound.Tag, ", restarting it: ", err) + } handled = false } if handled {