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
2026-09-30 12:29:09 +02:00
parent cada7a4442
commit 79fd705680
8 changed files with 699 additions and 11 deletions
--- 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 <key>` 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.

--- 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
 }
--- 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)},
 	}
 }
--- 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")))
+}
--- a/modules/lighthouse/channel.go
+++ 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
+}
--- a/modules/lighthouse/channel_test.go
+++ 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)
+	}
+}
--- 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 {
--- a/modules/lighthouse/registry.go
+++ 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
+}