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/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
 }

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)},
 	}
 }

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")))
+}