From 3b10e726b7c911d324408c9c65a8df172291051e Mon Sep 17 00:00:00 2001 From: Krispire Admin Date: Sat, 29 Aug 2026 01:11:22 +0530 Subject: [PATCH] feat: add private r2 bucket --- .env.example | 5 + internal/config/config.go | 8 + internal/handler/broadcasting_auth.go | 174 +++++++++++++++++++++ internal/handler/broadcasting_auth_test.go | 152 ++++++++++++++++++ internal/handler/soketi_webhook.go | 24 ++- internal/handler/soketi_webhook_test.go | 31 ++++ internal/middleware/auth.go | 7 + internal/repository/r2.go | 103 +++++++++++- internal/server/server.go | 19 +++ internal/service/notification_service.go | 10 +- pkg/realtime/auth.go | 119 ++++++++++++++ pkg/realtime/auth_test.go | 93 +++++++++++ 12 files changed, 731 insertions(+), 14 deletions(-) create mode 100644 internal/handler/broadcasting_auth.go create mode 100644 internal/handler/broadcasting_auth_test.go create mode 100644 internal/handler/soketi_webhook_test.go create mode 100644 pkg/realtime/auth.go create mode 100644 pkg/realtime/auth_test.go diff --git a/.env.example b/.env.example index b7f4cf8..4495b4c 100644 --- a/.env.example +++ b/.env.example @@ -24,6 +24,11 @@ R2_SECRET_ACCESS_KEY= R2_BUCKET= # Public base URL where bucket objects are served (custom domain or pub-XXX.r2.dev) R2_PUBLIC_URL= +# Separate bucket for private media (DM attachments), read only via short-lived +# presigned GETs. It must have NO public access and no custom domain: R2 public +# access is per-bucket, so a prefix inside R2_BUCKET would still be world-readable. +# Empty disables private-media endpoints (they return 503). +R2_PRIVATE_BUCKET= # SMTP (password-reset + email-verification + /feedback admin notifications) SMTP_HOST= diff --git a/internal/config/config.go b/internal/config/config.go index 15a9428..b833535 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -37,6 +37,13 @@ type Config struct { R2Bucket string R2PublicURL string // Public base URL for objects, e.g. https://cdn.example.com or the R2 public bucket URL + // R2PrivateBucket is a SEPARATE bucket with no public access, used for + // objects that must never be fetchable by URL alone (DM attachments). + // A prefix inside R2Bucket would not do: R2 public access is per-bucket, + // so anything in that bucket is world-readable. Empty disables the + // private-media endpoints (they return 503) — the BFF still boots. + R2PrivateBucket string + // SMTP — used for password-reset, email-verification, and // /feedback admin-notification mails. SMTPHost string @@ -136,6 +143,7 @@ func Load() (*Config, error) { R2SecretAccessKey: getEnv("R2_SECRET_ACCESS_KEY", ""), R2Bucket: getEnv("R2_BUCKET", ""), R2PublicURL: getEnv("R2_PUBLIC_URL", ""), + R2PrivateBucket: getEnv("R2_PRIVATE_BUCKET", ""), SMTPHost: getEnv("SMTP_HOST", ""), SMTPPort: getEnvInt("SMTP_PORT", 587), diff --git a/internal/handler/broadcasting_auth.go b/internal/handler/broadcasting_auth.go new file mode 100644 index 0000000..d26a819 --- /dev/null +++ b/internal/handler/broadcasting_auth.go @@ -0,0 +1,174 @@ +package handler + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "strings" + + "github.com/prabhupadasays/spsays-bff/internal/middleware" + "github.com/prabhupadasays/spsays-bff/pkg/realtime" + "github.com/prabhupadasays/spsays-bff/pkg/response" + "go.uber.org/zap" +) + +// ChannelRule authorizes one family of channels. +// +// Suffix is whatever follows Prefix in the channel name (a user ID today, a +// conversation ID once chat lands). Authorize reports whether userID — the +// authenticated caller — may subscribe. Returning an error means "couldn't +// decide" (a lookup failed) and surfaces as a 500; returning (false, nil) +// means "decided: no" and surfaces as a 403. +type ChannelRule struct { + Prefix string + Authorize func(ctx context.Context, userID, suffix string) (bool, error) +} + +// BroadcastingAuthHandler implements the Pusher channel-authorization +// endpoint that pusher-js calls (as `authEndpoint`) before subscribing to a +// private- or presence- channel. +// +// This is the gate that makes private channels private: Soketi will not honour +// a subscription without a signature, and the signature is only minted here, +// after RequireAuth has established who the caller is and a rule has agreed +// they may listen. Unknown channels are denied by default, so adding a channel +// family is an explicit act. +type BroadcastingAuthHandler struct { + rt *realtime.SoketiClient + rules []ChannelRule + logger *zap.Logger +} + +func NewBroadcastingAuthHandler(rt *realtime.SoketiClient, logger *zap.Logger, rules ...ChannelRule) *BroadcastingAuthHandler { + return &BroadcastingAuthHandler{rt: rt, rules: rules, logger: logger} +} + +// NotificationsRule allows a user to subscribe to their own notification +// channel and nobody else's. +func NotificationsRule() ChannelRule { + return ChannelRule{ + Prefix: "private-notifications.", + Authorize: func(_ context.Context, userID, suffix string) (bool, error) { + return suffix != "" && suffix == userID, nil + }, + } +} + +type broadcastingAuthResp struct { + Auth string `json:"auth"` + ChannelData string `json:"channel_data,omitempty"` +} + +// writeAuth emits the auth payload BARE, deliberately bypassing the API's +// {success, data} envelope. +// +// This endpoint speaks Pusher's protocol, not our API's: pusher-js reads +// `auth` off the top level of the parsed body, so an enveloped response makes +// every private subscription fail. Errors still use response.Error — the +// client only checks the status code on that path. +func writeAuth(w http.ResponseWriter, payload broadcastingAuthResp) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(payload) +} + +// Handle — POST /api/v1/broadcasting/auth. Session-authenticated (the cookie +// travels because the frontend sets credentials: "include"). +// +// pusher-js posts application/x-www-form-urlencoded by default; JSON is +// accepted too so a custom client isn't forced into form encoding. +func (h *BroadcastingAuthHandler) Handle(w http.ResponseWriter, r *http.Request) { + userID := middleware.UserID(r.Context()) + if userID == "" { + // RequireAuth should have caught this; refuse rather than sign. + response.Error(w, http.StatusUnauthorized, "authentication required") + return + } + + socketID, channel := parseBroadcastingAuthReq(r) + if socketID == "" || channel == "" { + response.Error(w, http.StatusBadRequest, "socket_id and channel_name are required") + return + } + + rule, suffix, found := h.match(channel) + if !found { + // Default-deny. An unrecognised channel is a client bug or an + // probe — either way there's nothing to sign. + h.logger.Warn("broadcasting auth: unknown channel", + zap.String("channel", channel), zap.String("user_id", userID)) + response.Error(w, http.StatusForbidden, "channel not available") + return + } + + allowed, err := rule.Authorize(r.Context(), userID, suffix) + if err != nil { + h.logger.Error("broadcasting auth: authorize failed", + zap.String("channel", channel), zap.Error(err)) + response.Error(w, http.StatusInternalServerError, "could not authorize channel") + return + } + if !allowed { + h.logger.Warn("broadcasting auth: denied", + zap.String("channel", channel), zap.String("user_id", userID)) + response.Error(w, http.StatusForbidden, "not permitted on this channel") + return + } + + if strings.HasPrefix(channel, "presence-") { + auth, channelData, err := h.rt.AuthorizePresence(socketID, channel, realtime.PresenceData{UserID: userID}) + if err != nil { + h.authError(w, err) + return + } + writeAuth(w, broadcastingAuthResp{Auth: auth, ChannelData: channelData}) + return + } + + auth, err := h.rt.AuthorizeChannel(socketID, channel) + if err != nil { + h.authError(w, err) + return + } + writeAuth(w, broadcastingAuthResp{Auth: auth}) +} + +// match finds the rule owning this channel and returns the part of the name +// after its prefix. +func (h *BroadcastingAuthHandler) match(channel string) (ChannelRule, string, bool) { + for _, rule := range h.rules { + if strings.HasPrefix(channel, rule.Prefix) { + return rule, strings.TrimPrefix(channel, rule.Prefix), true + } + } + return ChannelRule{}, "", false +} + +func (h *BroadcastingAuthHandler) authError(w http.ResponseWriter, err error) { + if errors.Is(err, realtime.ErrNotConfigured) { + response.Error(w, http.StatusServiceUnavailable, "realtime is not configured") + return + } + h.logger.Error("broadcasting auth: signing failed", zap.Error(err)) + response.Error(w, http.StatusBadRequest, err.Error()) +} + +// parseBroadcastingAuthReq reads socket_id / channel_name from either a form +// body (pusher-js default) or a JSON body. +func parseBroadcastingAuthReq(r *http.Request) (socketID, channel string) { + if strings.HasPrefix(r.Header.Get("Content-Type"), "application/json") { + var body struct { + SocketID string `json:"socket_id"` + ChannelName string `json:"channel_name"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + return "", "" + } + return strings.TrimSpace(body.SocketID), strings.TrimSpace(body.ChannelName) + } + if err := r.ParseForm(); err != nil { + return "", "" + } + return strings.TrimSpace(r.PostFormValue("socket_id")), strings.TrimSpace(r.PostFormValue("channel_name")) +} diff --git a/internal/handler/broadcasting_auth_test.go b/internal/handler/broadcasting_auth_test.go new file mode 100644 index 0000000..2ec5bf8 --- /dev/null +++ b/internal/handler/broadcasting_auth_test.go @@ -0,0 +1,152 @@ +package handler + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "testing" + + "github.com/prabhupadasays/spsays-bff/internal/middleware" + "github.com/prabhupadasays/spsays-bff/pkg/realtime" + "go.uber.org/zap" +) + +const testUser = "user-a" + +func testAuthHandler(t *testing.T) *BroadcastingAuthHandler { + t.Helper() + rt := realtime.NewSoketiClient("app-1", "test-key", "test-secret", "ws.example.com", 443, true, zap.NewNop()) + if rt == nil { + t.Fatal("realtime client is nil") + } + // A second rule stands in for the chat conversation channels, proving the + // rule table is what decides access rather than anything hardcoded. + convRule := ChannelRule{ + Prefix: "presence-conversation.", + Authorize: func(_ context.Context, userID, suffix string) (bool, error) { + return suffix == "open-to-user-a" && userID == testUser, nil + }, + } + return NewBroadcastingAuthHandler(rt, zap.NewNop(), NotificationsRule(), convRule) +} + +// serve posts a form-encoded auth request as userID ("" = unauthenticated). +func serve(t *testing.T, h *BroadcastingAuthHandler, userID, socketID, channel string) *httptest.ResponseRecorder { + t.Helper() + form := url.Values{"socket_id": {socketID}, "channel_name": {channel}} + req := httptest.NewRequest(http.MethodPost, "/api/v1/broadcasting/auth", strings.NewReader(form.Encode())) + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + if userID != "" { + req = req.WithContext(middleware.WithUserID(req.Context(), userID)) + } + rec := httptest.NewRecorder() + h.Handle(rec, req) + return rec +} + +func TestBroadcastingAuth_Authorization(t *testing.T) { + h := testAuthHandler(t) + + tests := []struct { + name string + userID string + channel string + want int + }{ + {"own notifications channel", testUser, "private-notifications." + testUser, http.StatusOK}, + {"another user's notifications", testUser, "private-notifications.user-b", http.StatusForbidden}, + {"unauthenticated", "", "private-notifications." + testUser, http.StatusUnauthorized}, + {"unknown channel family", testUser, "private-secrets.42", http.StatusForbidden}, + {"legacy public channel is not signable", testUser, "notifications." + testUser, http.StatusForbidden}, + {"empty suffix", testUser, "private-notifications.", http.StatusForbidden}, + {"presence channel allowed by rule", testUser, "presence-conversation.open-to-user-a", http.StatusOK}, + {"presence channel denied by rule", testUser, "presence-conversation.someone-elses", http.StatusForbidden}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + rec := serve(t, h, tc.userID, "1234.5678", tc.channel) + if rec.Code != tc.want { + t.Errorf("status = %d, want %d (body %s)", rec.Code, tc.want, rec.Body.String()) + } + }) + } +} + +// A prefix match must not be mistaken for authorization: "…user-a-evil" starts +// with the victim's id but is a different channel. +func TestBroadcastingAuth_SuffixIsComparedExactly(t *testing.T) { + rec := serve(t, testAuthHandler(t), testUser, "1234.5678", "private-notifications."+testUser+"-evil") + if rec.Code != http.StatusForbidden { + t.Errorf("status = %d, want 403", rec.Code) + } +} + +func TestBroadcastingAuth_SignatureShape(t *testing.T) { + rec := serve(t, testAuthHandler(t), testUser, "1234.5678", "private-notifications."+testUser) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rec.Code) + } + var body struct { + Auth string `json:"auth"` + ChannelData string `json:"channel_data"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil { + t.Fatalf("decode: %v", err) + } + if !strings.HasPrefix(body.Auth, "test-key:") { + t.Errorf("auth = %q, want a : pair", body.Auth) + } + // channel_data belongs to presence channels only; sending it for a private + // channel would make Soketi reject the subscription. + if body.ChannelData != "" { + t.Errorf("channel_data = %q, want empty for a private channel", body.ChannelData) + } +} + +func TestBroadcastingAuth_PresenceReturnsChannelData(t *testing.T) { + rec := serve(t, testAuthHandler(t), testUser, "1234.5678", "presence-conversation.open-to-user-a") + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rec.Code) + } + var body struct { + ChannelData string `json:"channel_data"` + } + _ = json.Unmarshal(rec.Body.Bytes(), &body) + if !strings.Contains(body.ChannelData, `"user_id":"`+testUser+`"`) { + t.Errorf("channel_data = %q, want the caller's user_id", body.ChannelData) + } +} + +func TestBroadcastingAuth_AcceptsJSONBody(t *testing.T) { + body := `{"socket_id":"1234.5678","channel_name":"private-notifications.` + testUser + `"}` + req := httptest.NewRequest(http.MethodPost, "/api/v1/broadcasting/auth", strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + req = req.WithContext(middleware.WithUserID(req.Context(), testUser)) + rec := httptest.NewRecorder() + testAuthHandler(t).Handle(rec, req) + if rec.Code != http.StatusOK { + t.Errorf("status = %d, want 200 (body %s)", rec.Code, rec.Body.String()) + } +} + +func TestBroadcastingAuth_MissingParams(t *testing.T) { + h := testAuthHandler(t) + if rec := serve(t, h, testUser, "", "private-notifications."+testUser); rec.Code != http.StatusBadRequest { + t.Errorf("missing socket_id: status = %d, want 400", rec.Code) + } + if rec := serve(t, h, testUser, "1234.5678", ""); rec.Code != http.StatusBadRequest { + t.Errorf("missing channel_name: status = %d, want 400", rec.Code) + } +} + +// With realtime disabled there is no secret to sign with; the endpoint must +// say so rather than hand back an unusable signature. +func TestBroadcastingAuth_UnconfiguredRealtime(t *testing.T) { + h := NewBroadcastingAuthHandler(nil, zap.NewNop(), NotificationsRule()) + if rec := serve(t, h, testUser, "1234.5678", "private-notifications."+testUser); rec.Code != http.StatusServiceUnavailable { + t.Errorf("status = %d, want 503", rec.Code) + } +} diff --git a/internal/handler/soketi_webhook.go b/internal/handler/soketi_webhook.go index 8233775..ee42b6d 100644 --- a/internal/handler/soketi_webhook.go +++ b/internal/handler/soketi_webhook.go @@ -10,14 +10,26 @@ import ( "strings" "github.com/prabhupadasays/spsays-bff/internal/repository" + "github.com/prabhupadasays/spsays-bff/pkg/realtime" "go.uber.org/zap" ) -// notifChannelPrefix is the public per-user channel the frontend subscribes to -// for realtime notifications. Because it's keyed by user, a `channel_vacated` -// webhook on it means that user's *last* connection dropped — i.e. they went -// offline (across all their tabs) — so we can evict their feed seen-set. -const notifChannelPrefix = "notifications." +// userIDFromNotifChannel extracts the user id from a per-user notifications +// channel, matching both the private name and the legacy public one. +// +// Because these channels are keyed by user, a `channel_vacated` webhook on one +// means that user's *last* connection dropped — i.e. they went offline across +// all their tabs — so we can evict their feed seen-set. Both names are matched +// for the duration of the private-channel migration: clients running a +// service-worker-cached bundle still sit on the legacy channel, and matching +// only one name would silently stop evicting for half the users. +func userIDFromNotifChannel(channel string) (string, bool) { + if uid, ok := strings.CutPrefix(channel, realtime.UserChannelPrefix); ok { + return uid, uid != "" + } + uid, ok := strings.CutPrefix(channel, realtime.LegacyUserChannelPrefix) + return uid, ok && uid != "" +} // SoketiWebhookHandler receives Soketi (Pusher-compatible) webhooks. Soketi // signs each delivery with the app secret (X-Pusher-Signature = hex @@ -83,7 +95,7 @@ func (h *SoketiWebhookHandler) Handle(w http.ResponseWriter, r *http.Request) { if e.Name != "channel_vacated" { continue } - if uid, ok := strings.CutPrefix(e.Channel, notifChannelPrefix); ok && uid != "" { + if uid, ok := userIDFromNotifChannel(e.Channel); ok { h.feed.Reset(ctx, uid) cleared++ } diff --git a/internal/handler/soketi_webhook_test.go b/internal/handler/soketi_webhook_test.go new file mode 100644 index 0000000..a4d83d6 --- /dev/null +++ b/internal/handler/soketi_webhook_test.go @@ -0,0 +1,31 @@ +package handler + +import "testing" + +// Both channel names must resolve during the private-channel migration: +// matching only one would silently stop evicting feed seen-sets for every +// client still on the other name. +func TestUserIDFromNotifChannel(t *testing.T) { + tests := []struct { + channel string + wantID string + wantOK bool + }{ + {"private-notifications.user-a", "user-a", true}, + {"notifications.user-a", "user-a", true}, + {"private-notifications.", "", false}, + {"notifications.", "", false}, + {"presence-conversation.c1", "", false}, + {"", "", false}, + // Guards against a naive suffix split: the private name must not be + // mistaken for the legacy one and yield a mangled id. + {"private-notifications.notifications.x", "notifications.x", true}, + } + for _, tc := range tests { + gotID, gotOK := userIDFromNotifChannel(tc.channel) + if gotOK != tc.wantOK || (tc.wantOK && gotID != tc.wantID) { + t.Errorf("userIDFromNotifChannel(%q) = (%q, %v), want (%q, %v)", + tc.channel, gotID, gotOK, tc.wantID, tc.wantOK) + } + } +} diff --git a/internal/middleware/auth.go b/internal/middleware/auth.go index 66c8786..f27a0da 100644 --- a/internal/middleware/auth.go +++ b/internal/middleware/auth.go @@ -100,6 +100,13 @@ func OptionalAuth(issuer *auth.Issuer, users *repository.UserRepo) func(http.Han } } +// WithUserID returns a context carrying userID, as RequireAuth would set it. +// The exported counterpart to UserID: it lets handlers be exercised in tests +// without standing up a JWT issuer and a user repository. +func WithUserID(ctx context.Context, userID string) context.Context { + return context.WithValue(ctx, userIDKey, userID) +} + // UserID returns the authenticated user's id from the request context. // Returns "" if RequireAuth has not run. func UserID(ctx context.Context) string { diff --git a/internal/repository/r2.go b/internal/repository/r2.go index 0cd6fb4..026cea9 100644 --- a/internal/repository/r2.go +++ b/internal/repository/r2.go @@ -25,6 +25,11 @@ type R2 struct { client *s3.Client bucket string publicBase string // R2_PUBLIC_URL, no trailing slash + + // privateBucket is a separate, non-public bucket for objects that must + // only be reachable through a presigned GET (see PresignGet). Empty when + // unconfigured, which makes PrivateConfigured report false. + privateBucket string } // R2Config captures the env vars needed to talk to R2. @@ -34,6 +39,9 @@ type R2Config struct { SecretAccessKey string Bucket string PublicURL string + // PrivateBucket is optional: absent means private media is disabled, + // while the public bucket keeps working. + PrivateBucket string } // Configured returns true if every required field is present. @@ -61,9 +69,10 @@ func NewR2(cfg R2Config) (*R2, error) { }) return &R2{ - client: client, - bucket: cfg.Bucket, - publicBase: trimTrailingSlash(cfg.PublicURL), + client: client, + bucket: cfg.Bucket, + publicBase: trimTrailingSlash(cfg.PublicURL), + privateBucket: cfg.PrivateBucket, }, nil } @@ -129,3 +138,91 @@ func trimTrailingSlash(s string) string { } return s } + +// --------------------------------------------------------------------------- +// Private media +// +// Objects in the public bucket are world-readable by URL — fine for avatars +// and post images, wrong for anything private (DM attachments). R2 public +// access is per-bucket, not per-prefix, so private objects live in a +// SEPARATE bucket that has no public access and no custom domain. The only +// way to read them is a short-lived presigned GET, which the BFF issues after +// checking that the caller is allowed to see that particular object. +// --------------------------------------------------------------------------- + +// PrivateConfigured reports whether the private bucket is available. Callers +// use it to return 503 rather than silently falling back to public storage — +// a fallback there would leak the very objects this bucket exists to protect. +func (r *R2) PrivateConfigured() bool { + return r != nil && r.privateBucket != "" +} + +// PresignPutPrivate returns a one-shot PUT URL for the PRIVATE bucket. +// +// Unlike PresignPut it deliberately returns no public URL: there isn't one, +// and handing callers a URL-shaped string for a private object invites it +// being persisted and treated as fetchable. Store the key instead, and mint a +// read URL per request with PresignGet. +func (r *R2) PresignPutPrivate(ctx context.Context, key, contentType string, expiry time.Duration) (uploadURL string, err error) { + if !r.PrivateConfigured() { + return "", errors.New("private media storage is not configured") + } + if key == "" { + return "", errors.New("empty key") + } + + presigner := s3.NewPresignClient(r.client) + req, err := presigner.PresignPutObject(ctx, &s3.PutObjectInput{ + Bucket: aws.String(r.privateBucket), + Key: aws.String(key), + ContentType: aws.String(contentType), + }, s3.WithPresignExpires(expiry)) + if err != nil { + return "", fmt.Errorf("presign private put: %w", err) + } + return req.URL, nil +} + +// PresignGet returns a short-lived read URL for an object in the private +// bucket. +// +// Keep expiry short. The URL is a bearer credential for that one object: +// anyone holding it can read until it expires, so it should be minted per +// request and never persisted in a message payload or an offline cache. +func (r *R2) PresignGet(ctx context.Context, key string, expiry time.Duration) (string, error) { + if !r.PrivateConfigured() { + return "", errors.New("private media storage is not configured") + } + if key == "" { + return "", errors.New("empty key") + } + + presigner := s3.NewPresignClient(r.client) + req, err := presigner.PresignGetObject(ctx, &s3.GetObjectInput{ + Bucket: aws.String(r.privateBucket), + Key: aws.String(key), + }, s3.WithPresignExpires(expiry)) + if err != nil { + return "", fmt.Errorf("presign get: %w", err) + } + return req.URL, nil +} + +// DeletePrivate removes an object from the private bucket. Retention and +// account deletion need real erasure, not a tombstone row that leaves the +// bytes sitting in storage. +func (r *R2) DeletePrivate(ctx context.Context, key string) error { + if !r.PrivateConfigured() { + return errors.New("private media storage is not configured") + } + if key == "" { + return errors.New("empty key") + } + if _, err := r.client.DeleteObject(ctx, &s3.DeleteObjectInput{ + Bucket: aws.String(r.privateBucket), + Key: aws.String(key), + }); err != nil { + return fmt.Errorf("delete private object %q: %w", key, err) + } + return nil +} diff --git a/internal/server/server.go b/internal/server/server.go index ecb18b3..30e3ae8 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -68,6 +68,7 @@ type Server struct { profileH *handler.ProfileHandler relationH *handler.RelationshipHandler soketiWebhookH *handler.SoketiWebhookHandler + broadcastAuthH *handler.BroadcastingAuthHandler notificationH *handler.NotificationHandler sadhnaH *handler.SadhnaHandler pushH *handler.PushHandler @@ -157,6 +158,7 @@ func (s *Server) initAuth(cfg *config.Config, logger *zap.Logger) { SecretAccessKey: cfg.R2SecretAccessKey, Bucket: cfg.R2Bucket, PublicURL: cfg.R2PublicURL, + PrivateBucket: cfg.R2PrivateBucket, }) if err != nil { logger.Warn("auth: R2 unavailable — avatar uploads will return 503", zap.Error(err)) @@ -256,6 +258,12 @@ func (s *Server) initAuth(cfg *config.Config, logger *zap.Logger) { cfg.SoketiAppID, cfg.SoketiAppKey, cfg.SoketiAppSecret, cfg.SoketiHost, cfg.SoketiPort, cfg.SoketiUseTLS, logger, ) + // Channel authorization for private-* subscriptions. Rules are explicit + // and default-deny; chat conversation channels get added here. + s.broadcastAuthH = handler.NewBroadcastingAuthHandler(rtClient, logger, + handler.NotificationsRule(), + ) + pushRepo := repository.NewPushRepo(s.pg) pushSender := service.NewPushSender(pushRepo, cfg.VAPIDPublicKey, cfg.VAPIDPrivateKey, cfg.VAPIDSubject, logger) s.pushH = handler.NewPushHandler(pushSender, pushRepo) @@ -887,6 +895,17 @@ func (s *Server) setupRouter() { r.Post("/webhooks/soketi", s.soketiWebhookH.Handle) } + // Pusher channel authorization — pusher-js posts here (as its + // `authEndpoint`) before subscribing to a private-/presence- channel. + // Session-authenticated: the signature is only minted for a caller + // who has proven who they are and passed the channel's rule. + if s.broadcastAuthH != nil { + r.Group(func(r chi.Router) { + r.Use(middleware.RequireAuth(s.authIssuer, s.userRepo)) + r.Post("/broadcasting/auth", s.broadcastAuthH.Handle) + }) + } + // Home feed — the global, personalised feed of posts + its "new on // the platform" discovery strip. Auth-required (the feed is // per-viewer: ordering + relationship flags depend on the caller). diff --git a/internal/service/notification_service.go b/internal/service/notification_service.go index d571d1a..86645c3 100644 --- a/internal/service/notification_service.go +++ b/internal/service/notification_service.go @@ -276,7 +276,7 @@ func (s *NotificationService) NotifyPostLike(ctx context.Context, postID, author payload["actor_handle"] = actorUser.Handle payload["actor_avatar_url"] = actorUser.AvatarURL } - if err := s.rt.Trigger(ctx, "notifications."+authorID, "notification", payload); err != nil { + if err := s.rt.TriggerUser(ctx, authorID, "notification", payload); err != nil { s.logger.Warn("realtime like publish failed", zap.String("recipient", authorID), zap.Error(err)) } } @@ -534,7 +534,7 @@ func (s *NotificationService) publishRealtime(ctx context.Context, recipientID, payload["actor_handle"] = actor.Handle payload["actor_avatar_url"] = actor.AvatarURL } - if err := s.rt.Trigger(ctx, "notifications."+recipientID, "notification", payload); err != nil { + if err := s.rt.TriggerUser(ctx, recipientID, "notification", payload); err != nil { s.logger.Warn("realtime notification publish failed", zap.String("recipient", recipientID), zap.String("kind", kind), zap.Error(err)) } @@ -550,7 +550,7 @@ func (s *NotificationService) SendReminder(ctx context.Context, userID, kind, ti // Realtime toast — a distinct "reminder" event so the client toasts // without bumping the unread badge (reminders have no DB row). if s.rt != nil && !prefOff(off, kind, repository.ChannelInApp) { - _ = s.rt.Trigger(ctx, "notifications."+userID, "reminder", map[string]any{ + _ = s.rt.TriggerUser(ctx, userID, "reminder", map[string]any{ "kind": kind, "title": title, "body": body, "url": url, }) } @@ -632,7 +632,7 @@ func (s *NotificationService) NotifyAdminMessage(ctx context.Context, recipientI url := s.frontendURL + "/notifications" if s.rt != nil { - if err := s.rt.Trigger(ctx, "notifications."+recipientID, "notification", map[string]any{ + if err := s.rt.TriggerUser(ctx, recipientID, "notification", map[string]any{ "kind": "admin_message", "body": body, }); err != nil { @@ -745,7 +745,7 @@ func (s *NotificationService) BroadcastDiscipleJoined(ctx context.Context, actor return nil } if s.rt != nil { - if err := s.rt.Trigger(ctx, "notifications."+uid, "notification", map[string]any{ + if err := s.rt.TriggerUser(ctx, uid, "notification", map[string]any{ "kind": NotifDiscipleJoined, "actor_name": name, "actor_handle": actor.Handle, diff --git a/pkg/realtime/auth.go b/pkg/realtime/auth.go new file mode 100644 index 0000000..ba9854a --- /dev/null +++ b/pkg/realtime/auth.go @@ -0,0 +1,119 @@ +package realtime + +import ( + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "strings" + + "go.uber.org/zap" +) + +// ErrNotConfigured is returned by the authorize helpers when realtime is +// disabled. Callers should surface it as a 503: without Soketi there is no +// subscription to authorize, and returning a bogus signature would just fail +// opaquely at the client. +var ErrNotConfigured = errors.New("realtime is not configured") + +// Channel-name prefixes, exported so consumers that parse channel names +// (the Soketi webhook) stay in step with the ones that build them. +const ( + UserChannelPrefix = "private-notifications." + LegacyUserChannelPrefix = "notifications." +) + +// UserChannel is the per-user private channel carrying notification nudges. +// Private (not public) so a subscription requires a signature the BFF only +// issues to that user — without it, anyone who learns a user's UUID can read +// their notification stream. +func UserChannel(userID string) string { return UserChannelPrefix + userID } + +// LegacyUserChannel is the public channel the frontend used before private +// channels existed. Still published to during the migration window so +// service-worker-cached bundles keep working; delete once clients have +// rolled over. See TriggerUser. +func LegacyUserChannel(userID string) string { return LegacyUserChannelPrefix + userID } + +// AuthorizeChannel signs a private-channel subscription. +// +// Pusher's scheme: the client sends its socket_id and the channel it wants; +// the app server — the only party holding the app secret — returns +// ":". Soketi verifies with +// its own copy of the secret. The signature is the whole authorization, so +// callers MUST check the caller may access the channel before calling this. +func (c *SoketiClient) AuthorizeChannel(socketID, channel string) (string, error) { + if c == nil { + return "", ErrNotConfigured + } + if socketID == "" || channel == "" { + return "", errors.New("socket_id and channel_name are required") + } + if strings.HasPrefix(channel, "presence-") { + return "", errors.New("presence channels must use AuthorizePresence") + } + return c.sign(socketID + ":" + channel), nil +} + +// PresenceData is the member identity Soketi publishes to everyone else on a +// presence channel. It is echoed to other subscribers verbatim, so UserInfo +// must carry only what those members may already see — never an email. +type PresenceData struct { + UserID string `json:"user_id"` + UserInfo map[string]any `json:"user_info,omitempty"` +} + +// AuthorizePresence signs a presence-channel subscription, returning both the +// auth signature and the channel_data the client must echo back. Presence +// folds channel_data into the signed string, so the member identity can't be +// tampered with client-side. +func (c *SoketiClient) AuthorizePresence(socketID, channel string, data PresenceData) (auth, channelData string, err error) { + if c == nil { + return "", "", ErrNotConfigured + } + if socketID == "" || channel == "" { + return "", "", errors.New("socket_id and channel_name are required") + } + if data.UserID == "" { + return "", "", errors.New("presence data requires a user_id") + } + encoded, err := json.Marshal(data) + if err != nil { + return "", "", fmt.Errorf("marshal channel_data: %w", err) + } + channelData = string(encoded) + return c.sign(socketID + ":" + channel + ":" + channelData), channelData, nil +} + +// sign returns the ":" auth string for an already-built +// string-to-sign. +func (c *SoketiClient) sign(stringToSign string) string { + mac := hmac.New(sha256.New, []byte(c.secret)) + mac.Write([]byte(stringToSign)) + return c.key + ":" + hex.EncodeToString(mac.Sum(nil)) +} + +// TriggerUser publishes a per-user event to both the private channel and the +// legacy public one. +// +// Dual-publish exists only for the migration: the frontend is a PWA, so a +// service-worker-cached bundle can keep subscribing to the public channel for +// a while after deploy, and a straight rename would silently stop its +// notifications. Once clients have rolled over, drop the legacy Trigger and +// this becomes a single publish. +// +// The private publish decides the error result; a legacy failure is not worth +// failing the caller over. +func (c *SoketiClient) TriggerUser(ctx context.Context, userID, event string, data any) error { + if c == nil { + return nil + } + err := c.Trigger(ctx, UserChannel(userID), event, data) + if legacyErr := c.Trigger(ctx, LegacyUserChannel(userID), event, data); legacyErr != nil { + c.logger.Debug("realtime: legacy channel publish failed", zap.Error(legacyErr)) + } + return err +} diff --git a/pkg/realtime/auth_test.go b/pkg/realtime/auth_test.go new file mode 100644 index 0000000..21ddae5 --- /dev/null +++ b/pkg/realtime/auth_test.go @@ -0,0 +1,93 @@ +package realtime + +import ( + "errors" + "strings" + "testing" + + "go.uber.org/zap" +) + +// testClient builds a client with fixed credentials so signatures are +// reproducible. +func testClient(t *testing.T) *SoketiClient { + t.Helper() + c := NewSoketiClient("app-1", "test-key", "test-secret", "ws.example.com", 443, true, zap.NewNop()) + if c == nil { + t.Fatal("NewSoketiClient returned nil for a fully configured client") + } + return c +} + +// The expected signatures are HMAC-SHA256 vectors computed independently of +// this package. Recomputing them with the same code under test would only +// prove the code equals itself; a fixed vector catches a change to the +// string-to-sign format, which is what Soketi actually verifies against. +func TestAuthorizeChannel_MatchesKnownVector(t *testing.T) { + auth, err := testClient(t).AuthorizeChannel("1234.5678", "private-notifications.user-a") + if err != nil { + t.Fatalf("AuthorizeChannel: %v", err) + } + const want = "test-key:9c436d58a9d030f68b35d636bb3e930c471d5c11edec6745c969e4eee1c3e3df" + if auth != want { + t.Errorf("auth = %q, want %q", auth, want) + } +} + +func TestAuthorizePresence_MatchesKnownVector(t *testing.T) { + auth, channelData, err := testClient(t).AuthorizePresence( + "1234.5678", "presence-conversation.c1", PresenceData{UserID: "user-a"}) + if err != nil { + t.Fatalf("AuthorizePresence: %v", err) + } + if channelData != `{"user_id":"user-a"}` { + t.Errorf("channel_data = %q", channelData) + } + const want = "test-key:cf5a222b506d46383cb4edac7707ba620e4ba2bbb83f64d1cf6463b4241efd24" + if auth != want { + t.Errorf("auth = %q, want %q", auth, want) + } +} + +// Presence signing folds channel_data into the signed string; routing a +// presence channel through AuthorizeChannel would produce a signature Soketi +// rejects, so it's refused outright rather than silently mis-signed. +func TestAuthorizeChannel_RejectsPresenceChannel(t *testing.T) { + if _, err := testClient(t).AuthorizeChannel("1234.5678", "presence-conversation.c1"); err == nil { + t.Fatal("expected an error for a presence channel") + } +} + +func TestAuthorize_NilClientReportsNotConfigured(t *testing.T) { + var c *SoketiClient + if _, err := c.AuthorizeChannel("1234.5678", "private-x"); !errors.Is(err, ErrNotConfigured) { + t.Errorf("AuthorizeChannel err = %v, want ErrNotConfigured", err) + } + if _, _, err := c.AuthorizePresence("1234.5678", "presence-x", PresenceData{UserID: "u"}); !errors.Is(err, ErrNotConfigured) { + t.Errorf("AuthorizePresence err = %v, want ErrNotConfigured", err) + } +} + +func TestAuthorize_RequiresArguments(t *testing.T) { + c := testClient(t) + if _, err := c.AuthorizeChannel("", "private-x"); err == nil { + t.Error("expected an error for an empty socket_id") + } + if _, err := c.AuthorizeChannel("1234.5678", ""); err == nil { + t.Error("expected an error for an empty channel") + } + if _, _, err := c.AuthorizePresence("1234.5678", "presence-x", PresenceData{}); err == nil { + t.Error("expected an error for presence data with no user_id") + } +} + +func TestUserChannels(t *testing.T) { + if got := UserChannel("u1"); got != "private-notifications.u1" { + t.Errorf("UserChannel = %q", got) + } + // The legacy name must stay public-prefixed: it exists so already-cached + // clients keep receiving events during the migration. + if got := LegacyUserChannel("u1"); got != "notifications.u1" || strings.HasPrefix(got, "private-") { + t.Errorf("LegacyUserChannel = %q", got) + } +}