summercms/modules/lighthouse/centrifugo/proxy_test.go

package centrifugo

import (
	"bytes"
	"context"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"strconv"
	"strings"
	"sync"
	"testing"

	"git.golem15.com/golem15/summercms/modules/backpack"
	"git.golem15.com/golem15/summercms/modules/lighthouse"
)

const (
	proxyTestSecret = "proxy-secret-test-only-4b1d"
	wrongSecret     = "not-the-proxy-secret-9c2e"
	denyBody        = `{"error":{"code":403,"message":"Access denied"}}`
	allowEmptyInfo  = `{"result":{"info":[]}}`
)

type lockedBuffer struct {
	mu  sync.Mutex
	buf bytes.Buffer
}

func (b *lockedBuffer) Write(p []byte) (int, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.buf.Write(p)
}

func (b *lockedBuffer) String() string {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.buf.String()
}

// acmeAuthorizer allows user 7 on acme:room:1 and records every call.
type acmeAuthorizer struct {
	mu    sync.Mutex
	calls []string
}

func (a *acmeAuthorizer) Authorize(ctx context.Context, userID uint, channel string) lighthouse.Result {
	a.mu.Lock()
	a.calls = append(a.calls, strings.Join([]string{uintString(userID), channel, lighthouse.ClientID(ctx)}, "|"))
	a.mu.Unlock()
	switch {
	case userID == 7 && (channel == "acme:room:1" || channel == "presence:acme:room:1"):
		return lighthouse.Allowed(nil)
	case userID == 7 && channel == "acme:room:info":
		return lighthouse.Allowed(map[string]any{"role": "owner"})
	case userID == 7 && channel == "acme:room:bad-info":
		return lighthouse.Allowed(map[string]any{"bad": make(chan int)})
	case userID == 7 && channel == "presence:acme:room:caps":
		r := lighthouse.Allowed(nil)
		r.Capabilities = []string{"prs", "sub"}
		r.Overrides = map[string]any{"join_leave": map[string]bool{"value": false}, "zeta": 1, "alpha": "a"}
		return r
	case channel == "acme:room:silent":
		return lighthouse.Result{}
	}
	return lighthouse.Denied("not a member of " + channel)
}

func (a *acmeAuthorizer) last() string {
	a.mu.Lock()
	defer a.mu.Unlock()
	if len(a.calls) == 0 {
		return ""
	}
	return a.calls[len(a.calls)-1]
}

func uintString(v uint) string { return strconv.FormatUint(uint64(v), 10) }

func proxyService(t *testing.T) (*lighthouse.Service, *acmeAuthorizer, *lockedBuffer) {
	t.Helper()
	app := backpack.New(nil)
	logs := &lockedBuffer{}
	if err := app.Publish(slog.New(slog.NewJSONHandler(logs, nil))); err != nil {
		t.Fatal(err)
	}
	svc, err := lighthouse.From(app)
	if err != nil {
		t.Fatal(err)
	}
	auth := &acmeAuthorizer{}
	if err := svc.Registry().Register("acme", auth); err != nil {
		t.Fatal(err)
	}
	return svc, auth, logs
}

func proxyCall(h http.HandlerFunc, secret *string, body string) *httptest.ResponseRecorder {
	req := httptest.NewRequest(http.MethodPost, "/api/realtime/subscribe", strings.NewReader(body))
	req.RemoteAddr = "203.0.113.9:4242"
	if secret != nil {
		req.Header.Set("X-Centrifugo-Secret", *secret)
	}
	rec := httptest.NewRecorder()
	h(rec, req)
	return rec
}

func strp(s string) *string { return &s }

// TestProxy covers RT-02, T-11-01 and T-11-02, porting the WinterCMS
// websockets security tests (WS-005, WS-007, WS-013): the proxy secret is
// compared in constant time and an empty configured secret denies
// everything; empty, zero and non-scalar users deny; channels are parsed
// with the presence and segment rules and routed byte-exactly to their
// namespace authorizer with the PHP (int) user id and the client id;
// allows answer the exact info, allow and override bytes; every deny is
// the same HTTP 200 body with the reason only in the logs, and no secret
// is ever logged.
func TestProxy(t *testing.T) {
	svc, auth, logs := proxyService(t)
	h := ProxyHandler(svc, Config{ProxySecret: proxyTestSecret})
	good := strp(proxyTestSecret)
	cases := []struct {
		name     string
		secret   *string
		body     string
		want     string
		reason   string
		authCall string
	}{
		{"missing_secret", nil, `{"user":"7","channel":"acme:room:1"}`, denyBody, "Invalid or missing proxy secret", ""},
		{"wrong_secret", strp(wrongSecret), `{"user":"7","channel":"acme:room:1"}`, denyBody, "Invalid or missing proxy secret", ""},
		{"secret_prefix", strp(proxyTestSecret[:10]), `{"user":"7","channel":"acme:room:1"}`, denyBody, "Invalid or missing proxy secret", ""},
		{"malformed_json", good, `{"user":`, denyBody, "Malformed proxy request", ""},
		{"empty_user", good, `{"user":"","channel":"acme:room:1"}`, denyBody, "Authentication required", ""},
		{"zero_user_string", good, `{"user":"0","channel":"acme:room:1"}`, denyBody, "Authentication required", ""},
		{"zero_user_number", good, `{"user":0,"channel":"acme:room:1"}`, denyBody, "Authentication required", ""},
		{"zero_user_float", good, `{"user":0.0,"channel":"acme:room:1"}`, denyBody, "Authentication required", ""},
		{"bool_user", good, `{"user":true,"channel":"acme:room:1"}`, denyBody, "Authentication required", ""},
		{"object_user", good, `{"user":{"id":7},"channel":"acme:room:1"}`, denyBody, "Authentication required", ""},
		{"missing_user", good, `{"channel":"acme:room:1"}`, denyBody, "Authentication required", ""},
		{"missing_channel", good, `{"user":"7"}`, denyBody, "Missing channel", ""},
		{"empty_channel", good, `{"user":"7","channel":""}`, denyBody, "Missing channel", ""},
		{"double_presence_ws005", good, `{"user":"7","channel":"presence:presence:acme:room:1"}`, denyBody, "Unknown channel namespace", ""},
		{"four_segments_ws005", good, `{"user":"7","channel":"acme:room:1:extra"}`, denyBody, "Unknown channel namespace", ""},
		{"leading_colon_ws005", good, `{"user":"7","channel":":acme:room"}`, denyBody, "Unknown channel namespace", ""},
		{"unknown_namespace", good, `{"user":"7","channel":"other:1"}`, denyBody, "Unknown channel namespace", ""},
		{"case_mismatched_namespace", good, `{"user":"7","channel":"ACME:room:1"}`, denyBody, "Unknown channel namespace", ""},
		{"allow_string_user", good, `{"user":"7","channel":"acme:room:1","client":"c-1"}`, allowEmptyInfo, "", "7|acme:room:1|c-1"},
		{"allow_number_user", good, `{"user":7,"channel":"acme:room:1"}`, allowEmptyInfo, "", "7|acme:room:1|"},
		{"php_int_cast_user", good, `{"user":"7abc","channel":"acme:room:1"}`, allowEmptyInfo, "", "7|acme:room:1|"},
		{"negative_user_is_zero", good, `{"user":"-5","channel":"acme:room:1"}`, denyBody, "not a member of acme:room:1", "0|acme:room:1|"},
		{"authorizer_deny", good, `{"user":"8","channel":"acme:room:1"}`, denyBody, "not a member of acme:room:1", "8|acme:room:1|"},
		{"authorizer_deny_without_reason", good, `{"user":"8","channel":"acme:room:silent"}`, denyBody, "Access denied", "8|acme:room:silent|"},
		{"allow_with_info", good, `{"user":"7","channel":"acme:room:info"}`, `{"result":{"info":{"role":"owner"}}}`, "", ""},
		{"unencodable_info_denies", good, `{"user":"7","channel":"acme:room:bad-info"}`, denyBody, "", ""},
		{"presence_defaults_ws013", good, `{"user":"7","channel":"presence:acme:room:1"}`,
			`{"result":{"info":[],"allow":["prs"],"override":{"presence":{"value":true},"join_leave":{"value":true},"force_push_join_leave":{"value":false}}}}`, "", "7|presence:acme:room:1|"},
		{"presence_override_merge", good, `{"user":"7","channel":"presence:acme:room:caps"}`,
			`{"result":{"info":[],"allow":["prs","sub"],"override":{"presence":{"value":true},"join_leave":{"value":false},"force_push_join_leave":{"value":false},"alpha":"a","zeta":1}}}`, "", ""},
		{"oversized_body", good, `{"user":"7","channel":"acme:room:1","pad":"` + strings.Repeat("x", 64<<10) + `"}`, denyBody, "Malformed proxy request", ""},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			before := auth.last()
			rec := proxyCall(h, c.secret, c.body)
			if rec.Code != http.StatusOK {
				t.Fatalf("status = %d, want 200 (Centrifugo reads non-200 as an internal error)", rec.Code)
			}
			if got := rec.Body.String(); got != c.want {
				t.Fatalf("body = %s\nwant %s", got, c.want)
			}
			if rec.Header().Get("Content-Type") != "application/json" || rec.Header().Get("Cache-Control") != "no-cache, private" {
				t.Fatalf("headers = %v", rec.Header())
			}
			if c.reason != "" && !strings.Contains(logs.String(), `"reason":"`+c.reason+`"`) {
				t.Fatalf("no deny log with reason %q:\n%s", c.reason, logs.String())
			}
			if c.authCall != "" && auth.last() != c.authCall {
				t.Fatalf("authorizer call = %q, want %q", auth.last(), c.authCall)
			}
			if c.authCall == "" && c.want == denyBody && c.reason != "" && !strings.HasPrefix(c.reason, "not a member") && c.reason != "Access denied" && auth.last() != before {
				t.Fatalf("authorizer was consulted for a request refused before it: %q", auth.last())
			}
		})
	}
	if out := logs.String(); strings.Contains(out, proxyTestSecret) || strings.Contains(out, wrongSecret) || strings.Contains(out, proxyTestSecret[:10]) {
		t.Fatalf("a secret reached the logs:\n%s", out)
	}
	if !strings.Contains(logs.String(), `"ip":"203.0.113.9"`) {
		t.Fatal("secret failures do not log the client IP")
	}

	t.Run("empty_configured_secret_denies_everything", func(t *testing.T) {
		off := ProxyHandler(svc, Config{})
		for _, secret := range []*string{nil, strp(""), strp(proxyTestSecret)} {
			if rec := proxyCall(off, secret, `{"user":"7","channel":"acme:room:1"}`); rec.Body.String() != denyBody {
				t.Fatalf("secret %v: body %s", secret, rec.Body.String())
			}
		}
	})

	t.Run("concurrent_subscribes", func(t *testing.T) {
		var wg sync.WaitGroup
		for i := range 16 {
			wg.Add(1)
			go func() {
				defer wg.Done()
				body, want := `{"user":"7","channel":"acme:room:1"}`, allowEmptyInfo
				if i%2 == 1 {
					body, want = `{"user":"8","channel":"acme:room:1"}`, denyBody
				}
				if rec := proxyCall(h, good, body); rec.Body.String() != want {
					t.Errorf("concurrent %d: %s", i, rec.Body.String())
				}
			}()
		}
		wg.Wait()
	})
}