From 79fd70568097becef9e53aa058e3016035a54d64 Mon Sep 17 00:00:00 2001 From: Jakub Zych Date: Wed, 30 Sep 2026 12:29:09 +0200 Subject: [PATCH] feat(11-03): re-authorize every Centrifugo subscribe through a namespace registry - lighthouse: Registry of namespace authorizers (Result, Allowed, Denied), ParseChannel, ChannelID with PHP (int)-cast semantics (PHPInt, pinned by a php -r table test), FormatChannels, WithClientID/ClientID - centrifugo: ProxyHandler (constant-time X-Centrifugo-Secret, HTTP 200 generic deny, info [] on allow, presence allow/override merge, 64 KiB body cap) mounted as the ServerToServer subscribe route - README: proxy contract, registry and channel rules --- modules/lighthouse/README.md | 40 +++- modules/lighthouse/centrifugo/config.go | 6 + modules/lighthouse/centrifugo/driver.go | 4 +- modules/lighthouse/centrifugo/handlers.go | 238 ++++++++++++++++++++++ modules/lighthouse/channel.go | 172 ++++++++++++++++ modules/lighthouse/channel_test.go | 120 +++++++++++ modules/lighthouse/lighthouse.go | 19 +- modules/lighthouse/registry.go | 111 ++++++++++ 8 files changed, 699 insertions(+), 11 deletions(-) create mode 100644 modules/lighthouse/channel.go create mode 100644 modules/lighthouse/channel_test.go create mode 100644 modules/lighthouse/registry.go diff --git a/modules/lighthouse/README.md b/modules/lighthouse/README.md index b69ebc1..40d10b3 100644 --- a/modules/lighthouse/README.md +++ b/modules/lighthouse/README.md @@ -14,7 +14,9 @@ lighthouse is the SummerCMS counterpart of the WinterCMS websockets plugin. The A driver may need HTTP endpoints, such as a token route for signed-in users or a callback the realtime server calls. It declares them as `lighthouse.Route` values, each tagged with a `lighthouse.Surface`. The application mounts them once with `lighthouse.Mount` and decides the guard, group and rate-limit bucket per surface. Switching drivers never edits the application's route file. -The `centrifugo` sub-package is the Centrifugo driver. It has a hand-rolled `net/http` client for the Centrifugo HTTP API, a token issuer with the claims of the WinterCMS `JwtTokenGenerator`, and the token route handler. +Channel authorization is transport-neutral too. Plugins register a `lighthouse.Authorizer` per channel namespace on the service's `lighthouse.Registry`. The driver's subscribe endpoint asks the authorizer of the channel's namespace on every subscribe, so a user who loses access is denied the next time the client subscribes. Nothing is cached. + +The `centrifugo` sub-package is the Centrifugo driver. It has a hand-rolled `net/http` client for the Centrifugo HTTP API, a token issuer with the claims of the WinterCMS `JwtTokenGenerator`, the token route handler, and the subscribe proxy handler. ## Features @@ -23,10 +25,23 @@ The `centrifugo` sub-package is the Centrifugo driver. It has a hand-rolled `net - The `lighthouse.Publisher` interface (`Publish` for one channel, `Broadcast` for several) and the `lighthouse.Driver` interface, which adds `Name` and `Routes`. - Route mounting by surface: `lighthouse.UserAuth`, `lighthouse.ServerToServer` and `lighthouse.Public`. `lighthouse.Mount` puts user and public routes in `Group` and server-to-server routes in `GroupRaw`, each with the surface middleware followed by `lighthouse.Surfaces.Middleware`. A `lighthouse.UserAuth` route with no user middleware is refused, so a token route can never be mounted without a guard. Every route is validated before any is registered. - Users and actors: the application installs a `lighthouse.UserLookup` with `lighthouse.Service.SetUserLookup`. `lighthouse.Service.User` loads a `lighthouse.User` (id and display name). `lighthouse.Service.Actor` returns the `lighthouse.Actor` of a request, and `lighthouse.SystemActor` when there is no signed-in user or the principal is a backend admin. +- Channel rules. A channel is `namespace:entity:id`, optionally prefixed once with `presence:`. + - `lighthouse.ParseChannel` returns the namespace. It returns "" for a `presence:presence:` prefix or for more than three segments. The lookup is byte-exact and case-sensitive. + - `lighthouse.ChannelID` returns segment 1 converted with PHP's `(int)` cast (`lighthouse.PHPInt`): `5abc` is 5, `abc` is 0, and out-of-range values saturate. + - `lighthouse.FormatChannels` lowercases channel names and applies the broadcast namespace prefix. +- Authorizer registry: `lighthouse.Registry` (from `lighthouse.Service.Registry`) maps namespaces to a `lighthouse.Authorizer` or `lighthouse.AuthorizerFunc`. Registering an empty namespace, a namespace that contains `:`, a nil authorizer or a namespace twice is an error. `lighthouse.Registry.Namespaces` is sorted. An authorizer returns `lighthouse.Allowed` (optionally with info, capabilities and overrides) or `lighthouse.Denied` with an internal reason that only reaches the logs. It reads the realtime client id with `lighthouse.ClientID`. - Centrifugo driver (`centrifugo.Driver`, driver name `centrifugo`): - `centrifugo.Client` POSTs `publish`, `broadcast`, `presence` and `unsubscribe` calls with `Authorization: apikey ` and a 5 s timeout. The publish body is `{"channel":…,"data":{"event":…,"payload":…,"timestamp":"…+00:00"}}`, with an empty payload sent as `[]`. Any 2xx status counts as success. With an empty API key nothing is sent and the call returns `centrifugo.ErrNotConfigured`. The key never appears in logs or errors. - `centrifugo.TokenIssuer` signs HS256 tokens with five generators, the same as the WinterCMS generator: `ForUser` (claims `sub`, `exp`, `info` with only `name`), `Subscription`, `Anonymous` (`sub` "" and a 5-minute lifetime), `ForIdentifier` (an empty `info` is encoded as `[]`) and `SubscriptionForIdentifier`. It refuses to sign with an empty secret. - `centrifugo.TokenHandler` serves the token route. It answers 401 `{"error":"Unauthorized"}` when no user is signed in, 503 `{"error":"WebSocket not configured"}` when the token secret is empty, and otherwise 200 `{"token":"…"}`. It sends `Cache-Control: no-cache, private` and no trailing newline. + - `centrifugo.ProxyHandler` is the subscribe proxy endpoint. Centrifugo reads a non-200 status as an internal error, so every answer is HTTP 200. The checks run in this order: + 1. `X-Centrifugo-Secret` must equal `realtime.centrifugo.proxy_secret`, compared in constant time. An empty configured secret denies every subscribe. + 2. An empty or `"0"` user denies. The user may arrive as a JSON string or number; any other type counts as empty. + 3. A missing channel denies. Centrifugo always sends one. + 4. The channel's namespace must have a registered authorizer. + 5. The authorizer receives the user id and the full original channel. + + An allow answers `{"result":{"info":…}}`, with an empty info encoded as `[]`. A `presence:` channel also gets `allow` (the authorizer's capabilities, or `["prs"]`) and `override`. The override starts from the defaults `presence` and `join_leave` true and `force_push_join_leave` false, then applies the authorizer's overrides. Every deny answers `{"error":{"code":403,"message":"Access denied"}}` and logs `Subscription denied` at Warn with the reason, and never either secret. The request body is capped at 64 KiB. ## Usage @@ -38,7 +53,7 @@ centrifugo: token_secret: "" # set with SUMMER_REALTIME__CENTRIFUGO__TOKEN_SECRET ``` -A plugin imports the driver package for its side effect, builds the service at Boot and installs a user lookup: +A plugin imports the driver package for its side effect, builds the service at Boot, installs a user lookup and registers its channel authorizers: ```go package acme @@ -60,7 +75,14 @@ func (p *Plugin) Boot(app *backpack.App) error { svc.SetUserLookup(func(ctx context.Context, id uint) (lighthouse.User, bool, error) { return lookupAcmeUser(ctx, id) // the application's own user model }) - return nil + // Allow room:{id} to members only; re-checked on every subscribe. + return svc.Registry().Register("room", lighthouse.AuthorizerFunc( + func(ctx context.Context, userID uint, channel string) lighthouse.Result { + if isRoomMember(ctx, userID, lighthouse.ChannelID(channel)) { + return lighthouse.Allowed(nil) + } + return lighthouse.Denied("not a room member") + })) } ``` @@ -92,7 +114,7 @@ for _, pub := range mem.Publications() { | Identifier | Description | |------------|-------------| | `lighthouse.From(app)` | The app's `*lighthouse.Service`, built and published on first use. | -| `lighthouse.Service` | The realtime service: `Driver`, `Logger`, `Namespace`, `Queue`, `Timeout`, `SetUserLookup`, `User`, `Actor`. | +| `lighthouse.Service` | The realtime service: `Driver`, `Registry`, `Logger`, `Namespace`, `Queue`, `Timeout`, `SetUserLookup`, `User`, `Actor`. | | `lighthouse.Publisher` | `Publish(ctx, channel, event, payload)` and `Broadcast(ctx, channels, event, payload)`. | | `lighthouse.Driver` | `lighthouse.Publisher` plus `Name()` and `Routes()`. | | `lighthouse.DriverFactory` | `func(app, svc) (lighthouse.Driver, error)`. | @@ -102,6 +124,11 @@ for _, pub := range mem.Publications() { | `lighthouse.Surface`, `lighthouse.UserAuth`, `lighthouse.ServerToServer`, `lighthouse.Public` | Who calls a route. | | `lighthouse.Surfaces` | Application middleware per surface plus `Middleware` for every route. | | `lighthouse.Mount(r, driver, surfaces)` | Registers a driver's routes. | +| `lighthouse.Authorizer`, `lighthouse.AuthorizerFunc` | `Authorize(ctx, userID, channel) lighthouse.Result`. | +| `lighthouse.Result`, `lighthouse.Allowed`, `lighthouse.Denied` | A subscribe decision: `Allowed`, `Info`, `Capabilities`, `Overrides` and `Reason()`. | +| `lighthouse.Registry`, `lighthouse.NewRegistry` | Namespace to authorizer map: `Register`, `Get`, `Namespaces`. | +| `lighthouse.ParseChannel`, `lighthouse.ChannelID`, `lighthouse.PHPInt`, `lighthouse.FormatChannels` | Channel rules. | +| `lighthouse.WithClientID`, `lighthouse.ClientID` | The realtime client id of a subscribe request, carried in the context. | | `lighthouse.User`, `lighthouse.UserLookup` | A user id with a display name, and the application's lookup. | | `lighthouse.Actor`, `lighthouse.SystemActor` | Who caused a broadcast: `{"user_id":…,"name":…}`. | | `lighthouse.DurationSetting(cfg, path)` | Reads a duration string or an integer number of seconds. | @@ -111,11 +138,12 @@ for _, pub := range mem.Publications() { | Identifier | Description | |------------|-------------| -| `centrifugo.Config`, `centrifugo.LoadConfig` | The `realtime.centrifugo.*` settings with their defaults. | +| `centrifugo.Config`, `centrifugo.LoadConfig` | The `realtime.centrifugo.*` settings with their defaults, plus `TrustedProxies` from `http.trusted_proxies` for logging client IPs. | | `centrifugo.Client`, `centrifugo.NewClient` | HTTP API client: `Publish`, `Broadcast`, `Presence`, `Unsubscribe`, `Enabled`, `DebugInfo`. | | `centrifugo.DebugInfo` | `api_url`, `enabled`, `api_key_set`. | | `centrifugo.TokenIssuer`, `centrifugo.NewTokenIssuer` | HS256 token generators: `ForUser`, `Subscription`, `Anonymous`, `ForIdentifier`, `SubscriptionForIdentifier`, `Configured`. | | `centrifugo.TokenHandler(svc, issuer)` | The token route handler. | +| `centrifugo.ProxyHandler(svc, cfg)` | The subscribe proxy handler. | | `centrifugo.Driver`, `centrifugo.NewDriver`, `centrifugo.DriverName` | The `lighthouse.Driver`, with `Client`, `Issuer` and `Config` accessors. | | `centrifugo.ErrNotConfigured` | Returned when the API key or token secret an operation needs is empty. | @@ -138,7 +166,7 @@ for _, pub := range mem.Publications() { ## Dependencies -- `backpack`, `bouncer`, `compass`, `pact` and `wire` from this repository. +- `backpack`, `bouncer`, `compass`, `pact` and `wire` from this repository; the centrifugo driver also uses `surf` for the client IP. - `github.com/golang-jwt/jwt/v5` (centrifugo token signing). - The Centrifugo client is plain `net/http`; no Centrifugo SDK is used. diff --git a/modules/lighthouse/centrifugo/config.go b/modules/lighthouse/centrifugo/config.go index 5dc3c06..f614d17 100644 --- a/modules/lighthouse/centrifugo/config.go +++ b/modules/lighthouse/centrifugo/config.go @@ -5,11 +5,13 @@ package centrifugo import ( + "net/netip" "strings" "time" "git.golem15.com/golem15/summercms/modules/compass" "git.golem15.com/golem15/summercms/modules/lighthouse" + "git.golem15.com/golem15/summercms/modules/surf" ) // Default values of the realtime.centrifugo.* keys. @@ -40,6 +42,9 @@ type Config struct { // TokenPath and SubscribePath are the mounted route paths. TokenPath string SubscribePath string + // TrustedProxies are the http.trusted_proxies used to log the client IP + // of a denied subscribe. + TrustedProxies []netip.Prefix } // LoadConfig reads realtime.centrifugo.* from c, filling the defaults. @@ -74,5 +79,6 @@ func LoadConfig(c *compass.Config) Config { if v := str("subscribe_path"); v != "" { cfg.SubscribePath = v } + cfg.TrustedProxies = surf.TrustedProxies(c) return cfg } diff --git a/modules/lighthouse/centrifugo/driver.go b/modules/lighthouse/centrifugo/driver.go index 8662bb2..cea644d 100644 --- a/modules/lighthouse/centrifugo/driver.go +++ b/modules/lighthouse/centrifugo/driver.go @@ -69,9 +69,11 @@ func (d *Driver) Broadcast(ctx context.Context, channels []string, event string, return d.client.Broadcast(ctx, channels, event, payload) } -// Routes returns GET token_path (UserAuth, TokenHandler). +// Routes returns GET token_path (UserAuth, TokenHandler) and POST +// subscribe_path (ServerToServer, ProxyHandler). func (d *Driver) Routes() []lighthouse.Route { return []lighthouse.Route{ {Name: "token", Method: http.MethodGet, Path: d.cfg.TokenPath, Surface: lighthouse.UserAuth, Handler: TokenHandler(d.svc, d.issuer)}, + {Name: "subscribe", Method: http.MethodPost, Path: d.cfg.SubscribePath, Surface: lighthouse.ServerToServer, Handler: ProxyHandler(d.svc, d.cfg)}, } } diff --git a/modules/lighthouse/centrifugo/handlers.go b/modules/lighthouse/centrifugo/handlers.go index f84e4be..ac8740d 100644 --- a/modules/lighthouse/centrifugo/handlers.go +++ b/modules/lighthouse/centrifugo/handlers.go @@ -2,11 +2,15 @@ package centrifugo import ( "bytes" + "crypto/subtle" "encoding/json" + "log/slog" "net/http" + "sort" "git.golem15.com/golem15/summercms/modules/bouncer" "git.golem15.com/golem15/summercms/modules/lighthouse" + "git.golem15.com/golem15/summercms/modules/surf" ) type errorBody struct { @@ -65,3 +69,237 @@ func writeJSON(w http.ResponseWriter, status int, v any) { w.WriteHeader(status) _, _ = w.Write(bytes.TrimSuffix(buf.Bytes(), []byte("\n"))) } + +// maxProxyBody caps a subscribe proxy request body. +const maxProxyBody = 64 << 10 + +// denyAnswer is the one answer every denied subscribe gets. Centrifugo +// reads a non-200 status as internal error 100, so a deny is HTTP 200 with +// an error body; it never reveals why. +var denyAnswer = json.RawMessage(`{"error":{"code":403,"message":"` + accessDenied + `"}}`) + +// accessDenied is the generic deny message, also logged when an authorizer +// gives no reason. +const accessDenied = "Access denied" + +// proxyRequest is the part of Centrifugo's subscribe proxy body the +// handler reads. +type proxyRequest struct { + User json.RawMessage `json:"user"` + Channel *string `json:"channel"` + Client string `json:"client"` +} + +// ProxyHandler is the Centrifugo subscribe proxy endpoint, the WinterCMS +// ProxyController::subscribe. Every subscribe is re-authorized: +// +// 1. X-Centrifugo-Secret must equal cfg.ProxySecret (constant-time; an +// empty configured secret denies everything); +// 2. an empty or "0" user denies; +// 3. the namespace comes from lighthouse.ParseChannel and must have a +// registered authorizer; +// 4. the authorizer gets the user id (PHP (int) cast, negative as 0) and +// the full original channel, with the client id in ctx. +// +// An allow answers {"result":{"info":…}} with info [] when empty; a +// presence channel also gets "allow" (the authorizer's capabilities or +// ["prs"]) and "override" (presence and join_leave true, +// force_push_join_leave false, then the authorizer's overrides). Every +// deny answers the same HTTP 200 body and logs the reason at Warn. +func ProxyHandler(svc *lighthouse.Service, cfg Config) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + log := svc.Logger() + provided := r.Header.Get("X-Centrifugo-Secret") + if cfg.ProxySecret == "" || subtle.ConstantTimeCompare([]byte(cfg.ProxySecret), []byte(provided)) != 1 { + deny(w, log, "Invalid or missing proxy secret", slog.String("ip", surf.ClientIP(r, cfg.TrustedProxies))) + return + } + var req proxyRequest + dec := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxProxyBody)) + dec.UseNumber() + if err := dec.Decode(&req); err != nil { + deny(w, log, "Malformed proxy request", slog.String("ip", surf.ClientIP(r, cfg.TrustedProxies))) + return + } + user := userInput(req.User) + channel := "" + if req.Channel != nil { + channel = *req.Channel + } + if user == "" || user == "0" { + deny(w, log, "Authentication required", slog.String("channel", channel), slog.String("client", req.Client)) + return + } + if req.Channel == nil || channel == "" { + // Centrifugo always sends a channel; WinterCMS would fail with + // a TypeError here. Denying is deliberate hardening. + deny(w, log, "Missing channel", slog.String("user", user), slog.String("client", req.Client)) + return + } + namespace, presence := lighthouse.ParseChannel(channel) + authorizer, ok := svc.Registry().Get(namespace) + if !ok { + deny(w, log, "Unknown channel namespace", + slog.String("namespace", namespace), slog.String("channel", channel), + slog.String("user", user), slog.String("client", req.Client)) + return + } + id := lighthouse.PHPInt(user) + if id < 0 { + id = 0 + } + ctx := lighthouse.WithClientID(r.Context(), req.Client) + res := authorizer.Authorize(ctx, uint(id), channel) + if !res.Allowed { + reason := res.Reason() + if reason == "" { + reason = accessDenied + } + deny(w, log, reason, + slog.String("user", user), slog.String("channel", channel), + slog.String("client", req.Client), slog.String("internal_reason", res.Reason())) + return + } + allow(w, res, presence) + } +} + +// userInput returns the user field as PHP's $request->input('user') string +// form: a JSON string as is, a JSON number as its literal. Anything else is +// "" (treated as empty). +func userInput(raw json.RawMessage) string { + raw = bytes.TrimSpace(raw) + if len(raw) == 0 { + return "" + } + switch raw[0] { + case '"': + var s string + if err := json.Unmarshal(raw, &s); err != nil { + return "" + } + return s + case '-', '0', '1', '2', '3', '4', '5', '6', '7', '8', '9': + var n json.Number + if err := json.Unmarshal(raw, &n); err != nil { + return "" + } + if f, err := n.Float64(); err == nil && f == 0 { + return "0" // PHP empty(0) and empty(0.0) + } + return n.String() + default: + return "" + } +} + +func deny(w http.ResponseWriter, log *slog.Logger, reason string, attrs ...slog.Attr) { + args := make([]any, 0, len(attrs)+1) + args = append(args, slog.String("reason", reason)) + for _, a := range attrs { + args = append(args, a) + } + log.Warn("Subscription denied", args...) + writeRaw(w, denyAnswer) +} + +func allow(w http.ResponseWriter, res lighthouse.Result, presence bool) { + var buf bytes.Buffer + buf.WriteString(`{"result":{"info":`) + if len(res.Info) == 0 { + buf.WriteString(`[]`) + } else if b, err := marshal(res.Info); err == nil { + buf.Write(b) + } else { + writeRaw(w, denyAnswer) + return + } + if presence { + caps := res.Capabilities + if caps == nil { + caps = []string{"prs"} + } + b, err := marshal(caps) + if err != nil { + writeRaw(w, denyAnswer) + return + } + buf.WriteString(`,"allow":`) + buf.Write(b) + o, err := overrideJSON(res.Overrides) + if err != nil { + writeRaw(w, denyAnswer) + return + } + buf.WriteString(`,"override":`) + buf.Write(o) + } + buf.WriteString(`}}`) + writeRaw(w, buf.Bytes()) +} + +// overrideJSON is PHP array_merge of the presence defaults with the +// authorizer's overrides: the default keys keep their order and take the +// authorizer's value when it names them; extra keys follow, sorted. +func overrideJSON(overrides map[string]any) ([]byte, error) { + defaults := []struct { + key string + value any + }{ + {"presence", map[string]bool{"value": true}}, + {"join_leave", map[string]bool{"value": true}}, + {"force_push_join_leave", map[string]bool{"value": false}}, + } + var buf bytes.Buffer + buf.WriteByte('{') + seen := map[string]bool{} + write := func(key string, value any) error { + k, err := marshal(key) + if err != nil { + return err + } + v, err := marshal(value) + if err != nil { + return err + } + if buf.Len() > 1 { + buf.WriteByte(',') + } + buf.Write(k) + buf.WriteByte(':') + buf.Write(v) + return nil + } + for _, d := range defaults { + v := d.value + if o, ok := overrides[d.key]; ok { + v = o + } + seen[d.key] = true + if err := write(d.key, v); err != nil { + return nil, err + } + } + extra := make([]string, 0, len(overrides)) + for k := range overrides { + if !seen[k] { + extra = append(extra, k) + } + } + sort.Strings(extra) + for _, k := range extra { + if err := write(k, overrides[k]); err != nil { + return nil, err + } + } + buf.WriteByte('}') + return buf.Bytes(), nil +} + +func writeRaw(w http.ResponseWriter, body []byte) { + h := w.Header() + h.Set("Content-Type", "application/json") + h.Set("Cache-Control", "no-cache, private") + w.WriteHeader(http.StatusOK) + _, _ = w.Write(bytes.TrimSuffix(body, []byte("\n"))) +} diff --git a/modules/lighthouse/channel.go b/modules/lighthouse/channel.go new file mode 100644 index 0000000..04613b9 --- /dev/null +++ b/modules/lighthouse/channel.go @@ -0,0 +1,172 @@ +package lighthouse + +import ( + "context" + "math" + "strconv" + "strings" +) + +// presencePrefix marks a presence channel. +const presencePrefix = "presence:" + +// ParseChannel returns the authorizer namespace of channel, the WinterCMS +// parseChannel rule: +// +// - a "presence:presence:" prefix yields "" (deny); +// - one "presence:" prefix is stripped, and presence is reported true; +// - more than three ":"-separated segments yield "" (deny); +// - otherwise the namespace is the first segment. +// +// The lookup is byte-exact: no case folding or Unicode normalization. +func ParseChannel(channel string) (namespace string, presence bool) { + presence = strings.HasPrefix(channel, presencePrefix) + if strings.HasPrefix(channel, presencePrefix+presencePrefix) { + return "", presence + } + name := strings.TrimPrefix(channel, presencePrefix) + parts := strings.Split(name, ":") + if len(parts) > 3 { + return "", presence + } + return parts[0], presence +} + +// ChannelID returns segment 1 of channel (0 when it is missing) converted +// with PHP's (int) cast, as the WinterCMS authorizers read +// explode(':', $channel)[1]. "5abc" is 5, "abc" is 0, "1e3" is 1000, and +// out-of-range values saturate. For "presence:collection:5" segment 1 is +// "collection", so the id is 0. +func ChannelID(channel string) int64 { + parts := strings.SplitN(channel, ":", 3) + if len(parts) < 2 { + return 0 + } + return PHPInt(parts[1]) +} + +// PHPInt converts s the way PHP 8's (int) cast converts a string: leading +// whitespace and a sign are allowed, the longest numeric prefix (digits, +// an optional fraction and exponent) is used, integers saturate at the +// int64 bounds, and non-finite floats become 0. +func PHPInt(s string) int64 { + i := 0 + for i < len(s) && isPHPSpace(s[i]) { + i++ + } + start := i + if i < len(s) && (s[i] == '+' || s[i] == '-') { + i++ + } + digitsStart := i + for i < len(s) && isDigit(s[i]) { + i++ + } + intDigits := i - digitsStart + isFloat := false + if i < len(s) && s[i] == '.' { + j := i + 1 + for j < len(s) && isDigit(s[j]) { + j++ + } + if intDigits > 0 || j > i+1 { + isFloat = true + i = j + } + } + if intDigits == 0 && !isFloat { + return 0 + } + if i < len(s) && (s[i] == 'e' || s[i] == 'E') { + j := i + 1 + if j < len(s) && (s[j] == '+' || s[j] == '-') { + j++ + } + k := j + for k < len(s) && isDigit(s[k]) { + k++ + } + if k > j { + isFloat = true + i = k + } + } + num := s[start:i] + if !isFloat { + n, err := strconv.ParseInt(num, 10, 64) + if err != nil { + if strings.HasPrefix(num, "-") { + return math.MinInt64 + } + return math.MaxInt64 + } + return n + } + f, _ := strconv.ParseFloat(num, 64) + switch { + case math.IsInf(f, 0) || math.IsNaN(f): + return 0 + case f >= math.MaxInt64: + return math.MaxInt64 + case f <= math.MinInt64: + return math.MinInt64 + default: + return int64(f) + } +} + +func isPHPSpace(c byte) bool { + return c == ' ' || c == '\t' || c == '\n' || c == '\r' || c == '\v' || c == '\f' +} + +func isDigit(c byte) bool { return c >= '0' && c <= '9' } + +// FormatChannels lowercases each channel (ASCII, as PHP strtolower) and +// prefixes lowercase(namespace)+":" unless the namespace is empty or the +// channel already starts with that prefix. +func FormatChannels(namespace string, channels []string) []string { + out := make([]string, len(channels)) + prefix := "" + if namespace != "" && namespace != "0" { + prefix = asciiLower(namespace) + ":" + } + for i, ch := range channels { + name := asciiLower(ch) + if prefix != "" && !strings.HasPrefix(name, prefix) { + name = prefix + name + } + out[i] = name + } + return out +} + +func asciiLower(s string) string { + b := []byte(s) + for i, c := range b { + if c >= 'A' && c <= 'Z' { + b[i] = c + ('a' - 'A') + } + } + return string(b) +} + +type clientIDKey struct{} + +// WithClientID returns ctx carrying the realtime client id of a subscribe +// request. Drivers call it before invoking an Authorizer. +func WithClientID(ctx context.Context, id string) context.Context { + if ctx == nil { + ctx = context.Background() + } + return context.WithValue(ctx, clientIDKey{}, id) +} + +// ClientID returns the realtime client id placed in ctx by the driver, or +// "" outside a subscribe request. +func ClientID(ctx context.Context) string { + if ctx == nil { + return "" + } + id, _ := ctx.Value(clientIDKey{}).(string) + return id +} diff --git a/modules/lighthouse/channel_test.go b/modules/lighthouse/channel_test.go new file mode 100644 index 0000000..48533fc --- /dev/null +++ b/modules/lighthouse/channel_test.go @@ -0,0 +1,120 @@ +package lighthouse + +import ( + "context" + "math" + "reflect" + "testing" +) + +// TestChannelIDMatchesPHP pins ChannelID to PHP's (int) cast of the id +// segment. The expected values were printed by PHP 8.5.10: +// +// php -r 'foreach ([...] as $s) var_dump((int)$s);' +func TestChannelIDMatchesPHP(t *testing.T) { + cases := []struct { + segment string + want int64 + }{ + {"5", 5}, {"05", 5}, {" 5", 5}, {"5abc", 5}, {"abc", 0}, {"-3", -3}, {"", 0}, + {"1e3", 1000}, {"0x1A", 0}, {"9999999999999999999", math.MaxInt64}, + {"\t5", 5}, {"\n5", 5}, {"5 ", 5}, {"+5", 5}, {" -7x", -7}, {"1.9", 1}, {"-0", 0}, + {"1e20", math.MaxInt64}, {"0b11", 0}, {"٥", 0}, {" ", 0}, + {"-9999999999999999999", math.MinInt64}, {"-1e20", math.MinInt64}, {"1e-3", 0}, + {".5", 0}, {"5.", 5}, {"1e", 1}, {"1e+2", 100}, {"\v5", 5}, {"\f5", 5}, {"\r5", 5}, + {"NaN", 0}, {"INF", 0}, {"1E3", 1000}, {"9223372036854775807", math.MaxInt64}, + {"9223372036854775808", math.MaxInt64}, {"1.5e3", 1500}, {"-.5e1", -5}, {"0005", 5}, + {"00", 0}, {"- 5", 0}, {"5e3abc", 5000}, {"12.7e1x", 127}, {"1e400", 0}, + } + for _, tc := range cases { + if got := ChannelID("collection:" + tc.segment); got != tc.want { + t.Errorf("ChannelID(collection:%q) = %d, want %d", tc.segment, got, tc.want) + } + } + if got := ChannelID("collection"); got != 0 { + t.Errorf("missing segment = %d, want 0", got) + } + if got := ChannelID("presence:collection:5"); got != 0 { + t.Errorf("presence:collection:5 = %d, want 0 (segment 1 is \"collection\")", got) + } + if got := ChannelID("wishlist:12:extra"); got != 12 { + t.Errorf("wishlist:12:extra = %d, want 12", got) + } +} + +func TestParseChannel(t *testing.T) { + cases := []struct { + channel string + namespace string + presence bool + }{ + {"collection:5", "collection", false}, + {"ns:entity:id", "ns", false}, + {"presence:acme:room:1", "acme", true}, + {"presence:collection:5", "collection", true}, + {"presence:presence:x", "", true}, + {"a:b:c:d", "", false}, + {"presence:a:b:c:d", "", true}, + {"", "", false}, + {":5", "", false}, + {"Collection:5", "Collection", false}, + } + for _, tc := range cases { + ns, presence := ParseChannel(tc.channel) + if ns != tc.namespace || presence != tc.presence { + t.Errorf("ParseChannel(%q) = (%q, %t), want (%q, %t)", tc.channel, ns, presence, tc.namespace, tc.presence) + } + } +} + +func TestFormatChannels(t *testing.T) { + got := FormatChannels("Acme", []string{"Collection:5", "acme:user:1", "ACME:x"}) + want := []string{"acme:collection:5", "acme:user:1", "acme:x"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("FormatChannels = %v, want %v", got, want) + } + if got := FormatChannels("", []string{"Collection:5"}); !reflect.DeepEqual(got, []string{"collection:5"}) { + t.Fatalf("no namespace = %v", got) + } +} + +func TestClientID(t *testing.T) { + if ClientID(context.Background()) != "" { + t.Fatal("empty ctx has a client id") + } + if got := ClientID(WithClientID(context.Background(), "c-1")); got != "c-1" { + t.Fatalf("ClientID = %q", got) + } +} + +func TestRegistry(t *testing.T) { + r := NewRegistry() + allow := AuthorizerFunc(func(context.Context, uint, string) Result { return Allowed(nil) }) + if err := r.Register("b", allow); err != nil { + t.Fatal(err) + } + if err := r.Register("a", allow); err != nil { + t.Fatal(err) + } + for _, bad := range []struct { + ns string + a Authorizer + }{{"", allow}, {"x:y", allow}, {"c", nil}, {"a", allow}} { + if err := r.Register(bad.ns, bad.a); err == nil { + t.Errorf("Register(%q) accepted", bad.ns) + } + } + if got := r.Namespaces(); !reflect.DeepEqual(got, []string{"a", "b"}) { + t.Fatalf("Namespaces = %v", got) + } + if _, ok := r.Get("A"); ok { + t.Fatal("lookup is not case-sensitive") + } + if a, ok := r.Get("a"); !ok || !a.Authorize(context.Background(), 1, "a:1").Allowed { + t.Fatal("Get(a) failed") + } + d := Denied("why") + if d.Allowed || d.Reason() != "why" { + t.Fatalf("Denied = %+v", d) + } +} diff --git a/modules/lighthouse/lighthouse.go b/modules/lighthouse/lighthouse.go index cf1904e..79e0ea9 100644 --- a/modules/lighthouse/lighthouse.go +++ b/modules/lighthouse/lighthouse.go @@ -33,6 +33,7 @@ const ( type Service struct { app *backpack.App driver Driver + registry *Registry log *slog.Logger namespace string queue string @@ -85,10 +86,11 @@ func From(app *backpack.App) (*Service, error) { func newService(app *backpack.App) *Service { svc := &Service{ - app: app, - log: loggerFromApp(app), - queue: DefaultQueue, - timeout: DefaultTimeout, + app: app, + registry: NewRegistry(), + log: loggerFromApp(app), + queue: DefaultQueue, + timeout: DefaultTimeout, } if app == nil || app.Config == nil { return svc @@ -112,6 +114,15 @@ func (s *Service) Driver() Driver { return s.driver } +// Registry returns the channel-namespace authorizer registry that the +// driver's subscribe authorization consults. +func (s *Service) Registry() *Registry { + if s == nil { + return nil + } + return s.registry +} + // Logger returns the app logger the service and its driver log through. func (s *Service) Logger() *slog.Logger { if s == nil || s.log == nil { diff --git a/modules/lighthouse/registry.go b/modules/lighthouse/registry.go new file mode 100644 index 0000000..8a7212e --- /dev/null +++ b/modules/lighthouse/registry.go @@ -0,0 +1,111 @@ +package lighthouse + +import ( + "context" + "fmt" + "sort" + "strings" + "sync" +) + +// Result is an authorizer's subscribe decision. Info, Capabilities and +// Overrides are passed to the realtime server on an allow; nil means "use +// the driver default". The deny reason is for logs only and never reaches +// the client. +type Result struct { + Allowed bool + Info map[string]any + Capabilities []string + Overrides map[string]any + reason string +} + +// Reason returns the internal deny reason ("" for an allow). +func (r Result) Reason() string { return r.reason } + +// Allowed returns an allow with info (nil for none). +func Allowed(info map[string]any) Result { + return Result{Allowed: true, Info: info} +} + +// Denied returns a deny with an internal reason for the logs. +func Denied(reason string) Result { + return Result{reason: reason} +} + +// Authorizer decides whether userID may subscribe to channel. It is called +// on every subscribe with the full original channel, including any +// "presence:" prefix, and must check current state (no caching). The +// driver's client id is available through ClientID(ctx). +type Authorizer interface { + Authorize(ctx context.Context, userID uint, channel string) Result +} + +// AuthorizerFunc adapts a function to Authorizer. +type AuthorizerFunc func(ctx context.Context, userID uint, channel string) Result + +// Authorize calls f. +func (f AuthorizerFunc) Authorize(ctx context.Context, userID uint, channel string) Result { + return f(ctx, userID, channel) +} + +// Registry maps channel namespaces to authorizers. It is safe for +// concurrent use. +type Registry struct { + mu sync.RWMutex + authorizers map[string]Authorizer +} + +// NewRegistry returns an empty registry. +func NewRegistry() *Registry { + return &Registry{authorizers: map[string]Authorizer{}} +} + +// Register adds the authorizer of namespace. An empty namespace, one that +// contains ":", a nil authorizer, or a namespace registered twice is an +// error (unlike the WinterCMS registry, a second registration does not +// silently replace the first). +func (r *Registry) Register(namespace string, a Authorizer) error { + if namespace == "" { + return fmt.Errorf("lighthouse: authorizer namespace is empty") + } + if strings.Contains(namespace, ":") { + return fmt.Errorf("lighthouse: authorizer namespace %q contains \":\"", namespace) + } + if a == nil { + return fmt.Errorf("lighthouse: authorizer for namespace %q is nil", namespace) + } + r.mu.Lock() + defer r.mu.Unlock() + if _, dup := r.authorizers[namespace]; dup { + return fmt.Errorf("lighthouse: authorizer namespace %q is already registered", namespace) + } + r.authorizers[namespace] = a + return nil +} + +// Get returns the authorizer of namespace. The lookup is byte-exact. +func (r *Registry) Get(namespace string) (Authorizer, bool) { + if r == nil { + return nil, false + } + r.mu.RLock() + defer r.mu.RUnlock() + a, ok := r.authorizers[namespace] + return a, ok +} + +// Namespaces returns the registered namespaces, sorted. +func (r *Registry) Namespaces() []string { + if r == nil { + return nil + } + r.mu.RLock() + defer r.mu.RUnlock() + out := make([]string, 0, len(r.authorizers)) + for ns := range r.authorizers { + out = append(out, ns) + } + sort.Strings(out) + return out +}