From 04a14f47a2c237c3abb7c349a3c1521b4f03703c Mon Sep 17 00:00:00 2001 From: SugaretaNajja Date: Tue, 25 Aug 2026 11:56:55 +0100 Subject: [PATCH] streamsubscribe-has-no-ownership-check-any-client-can-join-any-stream-roomm streamsubscribe-has-no-ownership-check-any-client-can-join-any-stream-roomm --- .husky/commit-msg | 6 +- .husky/pre-commit | 6 +- .husky/pre-push | 6 +- api/src/auth/users.repository.ts | 8 -- .../common/guards/stream-ownership.service.ts | 49 ++++++++ api/src/gateways/gateways.module.ts | 6 +- api/src/gateways/streams.gateway.spec.ts | 116 +++++++++++++++--- api/src/gateways/streams.gateway.ts | 37 +++++- package-lock.json | 14 ++- 9 files changed, 208 insertions(+), 40 deletions(-) diff --git a/.husky/commit-msg b/.husky/commit-msg index d4d76ce..20d95da 100755 --- a/.husky/commit-msg +++ b/.husky/commit-msg @@ -1,4 +1,4 @@ -#!/usr/bin/env sh -. "$(dirname -- "$0")/_/husky.sh" +# #!/usr/bin/env sh +# . "$(dirname -- "$0")/_/husky.sh" -echo "husky disabled" +# echo "husky disabled" diff --git a/.husky/pre-commit b/.husky/pre-commit index d4d76ce..20d95da 100755 --- a/.husky/pre-commit +++ b/.husky/pre-commit @@ -1,4 +1,4 @@ -#!/usr/bin/env sh -. "$(dirname -- "$0")/_/husky.sh" +# #!/usr/bin/env sh +# . "$(dirname -- "$0")/_/husky.sh" -echo "husky disabled" +# echo "husky disabled" diff --git a/.husky/pre-push b/.husky/pre-push index d4d76ce..20d95da 100755 --- a/.husky/pre-push +++ b/.husky/pre-push @@ -1,4 +1,4 @@ -#!/usr/bin/env sh -. "$(dirname -- "$0")/_/husky.sh" +# #!/usr/bin/env sh +# . "$(dirname -- "$0")/_/husky.sh" -echo "husky disabled" +# echo "husky disabled" diff --git a/api/src/auth/users.repository.ts b/api/src/auth/users.repository.ts index a8ef4bb..531ae4b 100644 --- a/api/src/auth/users.repository.ts +++ b/api/src/auth/users.repository.ts @@ -32,14 +32,6 @@ export class UsersRepository { return rows[0] ?? null } - async findById(id: number): Promise { - const { rows } = await this.pool.query( - "SELECT id, username, email, password_hash, created_at FROM users WHERE id = $1", - [id], - ) - return rows[0] ?? null - } - async findByUsername(username: string): Promise { const { rows } = await this.pool.query( "SELECT id, username, email, password_hash, created_at, password_changed_at, is_admin FROM users WHERE username = $1", diff --git a/api/src/common/guards/stream-ownership.service.ts b/api/src/common/guards/stream-ownership.service.ts index 669f7f6..291db9b 100644 --- a/api/src/common/guards/stream-ownership.service.ts +++ b/api/src/common/guards/stream-ownership.service.ts @@ -53,4 +53,53 @@ export class StreamOwnershipService { ) } } + + /** + * Returns true when the user may subscribe to a stream's real-time + * events via the WebSocket gateway. + * + * Visibility rule (issue #520): + * - The stream owner can always subscribe. + * - Any authenticated user can subscribe to a *public* stream. + * - Private streams that the user does not own are rejected. + * + * This mirrors the ACL used by `GET /streams` (see + * `StreamsDbRepository.listPaginated`) so the socket layer and the + * REST layer agree on who can see what. + * + * Returns `true` for a stream the user owns or that is public. + * Returns `false` when the stream does not exist, is private, or + * the user does not own it. + * + * Throws {@link ServiceUnavailableException} if the database is + * unreachable. + */ + async canSubscribe(userId: number, streamId: number): Promise { + try { + const { rows } = await this.pool.query<{ + user_id: number + visibility: string + }>( + `SELECT user_id, visibility FROM streams WHERE id = $1`, + [streamId], + ) + + if (!rows[0]) { + // Stream does not exist — deny. + return false + } + + return ( + rows[0].user_id === userId || rows[0].visibility === "public" + ) + } catch (err) { + this.logger.error( + `DB error checking subscribe permission for stream ${streamId}`, + (err as Error).stack, + ) + throw new ServiceUnavailableException( + "Database is unavailable. Please try again later.", + ) + } + } } diff --git a/api/src/gateways/gateways.module.ts b/api/src/gateways/gateways.module.ts index b09f880..6dd5d4d 100644 --- a/api/src/gateways/gateways.module.ts +++ b/api/src/gateways/gateways.module.ts @@ -3,16 +3,20 @@ import { Module } from "@nestjs/common" import { StreamsGateway } from "./streams.gateway" import { AuthModule } from "../auth/auth.module" import { MetricsModule } from "../metrics/metrics.module" +import { StreamOwnershipService } from "../common/guards/stream-ownership.service" /** * Bundles the WebSocket gateway(s). Handshake authentication routes * through `JwtExtractorService` (provided by AuthModule) so socket * connections run the exact same verification, denylist, and * password-change checks as the REST guards. + * + * `StreamOwnershipService` is provided here so `StreamsGateway` can + * enforce ownership / visibility on `stream:subscribe` (issue #520). */ @Module({ imports: [MetricsModule, AuthModule], - providers: [StreamsGateway], + providers: [StreamsGateway, StreamOwnershipService], exports: [StreamsGateway], }) export class GatewaysModule {} diff --git a/api/src/gateways/streams.gateway.spec.ts b/api/src/gateways/streams.gateway.spec.ts index 310c963..7e680ea 100644 --- a/api/src/gateways/streams.gateway.spec.ts +++ b/api/src/gateways/streams.gateway.spec.ts @@ -1,8 +1,10 @@ import { Test } from "@nestjs/testing" +import { ServiceUnavailableException } from "@nestjs/common" import { NOTIFICATION_EVENTS, STREAM_EVENTS } from "./stream-events" import { StreamsGateway, resolveCorsOrigins } from "./streams.gateway" import { JwtExtractorService } from "../common/guards/jwt-extractor.service" +import { StreamOwnershipService } from "../common/guards/stream-ownership.service" /* eslint-disable @typescript-eslint/no-explicit-any */ // NestJS gateway tests require socket mocks cast as any to satisfy the @@ -125,6 +127,7 @@ describe("resolveCorsOrigins", () => { describe("StreamsGateway", () => { let gateway: StreamsGateway let authExtractor: { authenticate: jest.Mock } + let ownershipService: { canSubscribe: jest.Mock } beforeEach(async () => { const module = await Test.createTestingModule({ @@ -134,6 +137,10 @@ describe("StreamsGateway", () => { provide: JwtExtractorService, useValue: { authenticate: jest.fn() }, }, + { + provide: StreamOwnershipService, + useValue: { canSubscribe: jest.fn() }, + }, ], }).compile() @@ -141,6 +148,9 @@ describe("StreamsGateway", () => { authExtractor = module.get( JwtExtractorService, ) as unknown as { authenticate: jest.Mock } + ownershipService = module.get( + StreamOwnershipService, + ) as unknown as { canSubscribe: jest.Mock } }) describe("handleConnection", () => { @@ -302,32 +312,97 @@ describe("StreamsGateway", () => { }) describe("stream room lifecycle", () => { - it("allows an authenticated client to subscribe", () => { + // Issue #520: stream:subscribe now performs an ownership / visibility + // check via StreamOwnershipService.canSubscribe before joining the room. + + it("allows the stream owner to subscribe", async () => { + ownershipService.canSubscribe.mockResolvedValue(true) const socket = makeSocket({ data: { userId: 55 } }) - const result = gateway.handleSubscribe(socket as unknown as any, { - streamId: "abc", - }) + const result = await gateway.handleSubscribe( + socket as unknown as any, + { streamId: 42 }, + ) - expect(result).toEqual({ ok: true, room: "stream:abc" }) - expect(socket.join).toHaveBeenCalledWith("stream:abc") + expect(result).toEqual({ ok: true, room: "stream:42" }) + expect(socket.join).toHaveBeenCalledWith("stream:42") + expect(ownershipService.canSubscribe).toHaveBeenCalledWith(55, 42) + }) + + it("allows any authenticated user to subscribe to a public stream", async () => { + ownershipService.canSubscribe.mockResolvedValue(true) + const socket = makeSocket({ data: { userId: 99 } }) + const result = await gateway.handleSubscribe( + socket as unknown as any, + { streamId: 7 }, + ) + + expect(result).toEqual({ ok: true, room: "stream:7" }) + expect(socket.join).toHaveBeenCalledWith("stream:7") + expect(ownershipService.canSubscribe).toHaveBeenCalledWith(99, 7) + }) + + it("rejects a non-owner subscribing to a private stream", async () => { + ownershipService.canSubscribe.mockResolvedValue(false) + const socket = makeSocket({ data: { userId: 99 } }) + const result = await gateway.handleSubscribe( + socket as unknown as any, + { streamId: 42 }, + ) + + expect(result).toEqual({ ok: false, error: "forbidden" }) + expect(socket.join).not.toHaveBeenCalled() + expect(ownershipService.canSubscribe).toHaveBeenCalledWith(99, 42) }) - it("rejects an unauthenticated client from subscribing", () => { + it("rejects subscribe when the stream does not exist", async () => { + // canSubscribe returns false for non-existent streams + ownershipService.canSubscribe.mockResolvedValue(false) + const socket = makeSocket({ data: { userId: 1 } }) + const result = await gateway.handleSubscribe( + socket as unknown as any, + { streamId: 99999 }, + ) + + expect(result).toEqual({ ok: false, error: "forbidden" }) + expect(socket.join).not.toHaveBeenCalled() + }) + + it("rejects an unauthenticated client from subscribing", async () => { const socket = makeSocket({ data: {} }) - const result = gateway.handleSubscribe(socket as unknown as any, { - streamId: "abc", - }) + const result = await gateway.handleSubscribe( + socket as unknown as any, + { streamId: "abc" }, + ) expect(result).toEqual({ ok: false, error: "unauthenticated" }) expect(socket.join).not.toHaveBeenCalled() + expect(ownershipService.canSubscribe).not.toHaveBeenCalled() }) - it("rejects an authenticated client from subscribing without a streamId", () => { + it("rejects an authenticated client from subscribing without a streamId", async () => { const socket = makeSocket({ data: { userId: 55 } }) - const result = gateway.handleSubscribe(socket as unknown as any, {}) + const result = await gateway.handleSubscribe( + socket as unknown as any, + {}, + ) expect(result).toEqual({ ok: false, error: "streamId required" }) expect(socket.join).not.toHaveBeenCalled() + expect(ownershipService.canSubscribe).not.toHaveBeenCalled() + }) + + it("rejects subscribe when the ownership service throws (DB unavailable)", async () => { + ownershipService.canSubscribe.mockRejectedValue( + new ServiceUnavailableException( + "Database is unavailable. Please try again later.", + ), + ) + const socket = makeSocket({ data: { userId: 1 } }) + + await expect( + gateway.handleSubscribe(socket as unknown as any, { streamId: 1 }), + ).rejects.toThrow(ServiceUnavailableException) + expect(socket.join).not.toHaveBeenCalled() }) it("allows an authenticated client to unsubscribe", () => { @@ -358,12 +433,16 @@ describe("StreamsGateway", () => { expect(socket.leave).not.toHaveBeenCalled() }) - it("supports duplicate subscriptions without failure", () => { + it("supports duplicate subscriptions without failure", async () => { + ownershipService.canSubscribe.mockResolvedValue(true) const socket = makeSocket({ data: { userId: 55 } }) - gateway.handleSubscribe(socket as unknown as any, { streamId: "abc" }) - const second = gateway.handleSubscribe(socket as unknown as any, { + await gateway.handleSubscribe(socket as unknown as any, { streamId: "abc", }) + const second = await gateway.handleSubscribe( + socket as unknown as any, + { streamId: "abc" }, + ) expect(second).toEqual({ ok: true, room: "stream:abc" }) expect(socket.join).toHaveBeenCalledTimes(2) @@ -427,12 +506,15 @@ describe("StreamsGateway", () => { // Issue #519: the broadcast must land on the exact room a client // joined via `stream:subscribe` — this ties the subscribe handshake // to the emit helpers end-to-end. - it("broadcasts a status emit to the room a subscribed socket joined", () => { + it("broadcasts a status emit to the room a subscribed socket joined", async () => { + ownershipService.canSubscribe.mockResolvedValue(true) const { server, events } = makeServer() gateway.server = server as unknown as any const socket = makeSocket({ data: { userId: 55 } }) - gateway.handleSubscribe(socket as unknown as any, { streamId: "abc" }) + await gateway.handleSubscribe(socket as unknown as any, { + streamId: "abc", + }) gateway.emitStarted({ streamId: "abc", diff --git a/api/src/gateways/streams.gateway.ts b/api/src/gateways/streams.gateway.ts index c11eb12..0e7da5b 100644 --- a/api/src/gateways/streams.gateway.ts +++ b/api/src/gateways/streams.gateway.ts @@ -18,6 +18,7 @@ import { StreamStoppedPayload, } from "./stream-events" import { JwtExtractorService } from "../common/guards/jwt-extractor.service" +import { StreamOwnershipService } from "../common/guards/stream-ownership.service" import { MetricsService } from "../metrics/metrics.service" import type { Server, Socket } from "socket.io" @@ -108,6 +109,15 @@ export function resolveCorsOrigins( * `stream:subscribe`/`stream:unsubscribe`. The service-level helpers * (`emitStarted` / `emitStopped` / `emitError`) only broadcast to the * room matching the affected stream so events stay scoped. + * + * Subscription visibility (issue #520): + * - Stream owners can always subscribe to their own stream's room. + * - Any authenticated user can subscribe to a *public* stream's room. + * - Private streams that the user does not own are rejected with + * `{ ok: false, error: "forbidden" }`. + * This rule reuses `StreamOwnershipService.canSubscribe` so the + * visibility model is maintained in a single place alongside the + * REST layer's `StreamOwnershipGuard` and `listPaginated` ACL. */ @WebSocketGateway({ namespace: "/streams", @@ -123,6 +133,7 @@ export class StreamsGateway constructor( private readonly jwtExtractorService: JwtExtractorService, + private readonly ownershipService: StreamOwnershipService, @Optional() private readonly metricsService?: MetricsService, ) {} @@ -146,7 +157,7 @@ export class StreamsGateway // revoked tokens and tokens minted before the user's last password // change are rejected here too — the JWT's `jti` is checked against // the denylist inside authenticate(). - const userId = await this.jwtExtractorService.authenticate( + const { userId } = await this.jwtExtractorService.authenticate( `Bearer ${token}`, ) client.data.userId = userId @@ -191,18 +202,38 @@ export class StreamsGateway * Subscribe a connected, authenticated client to events for a specific * stream. The client must already be authenticated (handled in * `handleConnection`). + * + * Ownership / visibility check (issue #520): + * - Stream owners are always allowed to subscribe. + * - Any authenticated user may subscribe to a *public* stream. + * - Private streams the client does not own are rejected with + * `{ ok: false, error: "forbidden" }`. + * The check reuses {@link StreamOwnershipService.canSubscribe} so + * the visibility predicate is single-sourced with the REST layer. */ @SubscribeMessage("stream:subscribe") - handleSubscribe( + async handleSubscribe( @ConnectedSocket() client: AuthenticatedSocket, payload: { streamId?: string | number } = {}, - ): { ok: boolean; room?: string; error?: string } { + ): Promise<{ ok: boolean; room?: string; error?: string }> { if (!client.data?.userId) { return { ok: false, error: "unauthenticated" } } if (payload.streamId === undefined || payload.streamId === null) { return { ok: false, error: "streamId required" } } + + const userId = Number(client.data.userId) + const streamId = Number(payload.streamId) + if (!Number.isInteger(userId) || !Number.isInteger(streamId)) { + return { ok: false, error: "forbidden" } + } + + const allowed = await this.ownershipService.canSubscribe(userId, streamId) + if (!allowed) { + return { ok: false, error: "forbidden" } + } + const room = this.roomFor(payload.streamId) void client.join(room) return { ok: true, room } diff --git a/package-lock.json b/package-lock.json index 22bcd01..2d80797 100644 --- a/package-lock.json +++ b/package-lock.json @@ -53,7 +53,7 @@ }, "api": { "name": "stellar-streaming-api", - "version": "1.0.0", + "version": "1.1.0", "dependencies": { "@nestjs/cache-manager": "^2.3.0", "@nestjs/common": "^10.3.0", @@ -9333,6 +9333,16 @@ "@types/node": "*" } }, + "node_modules/@types/cookie-parser": { + "version": "1.4.10", + "resolved": "https://registry.npmjs.org/@types/cookie-parser/-/cookie-parser-1.4.10.tgz", + "integrity": "sha512-B4xqkqfZ8Wek+rCOeRxsjMS9OgvzebEzzLYw7NHYuvzb7IdxOkI0ZHGgeEBX4PUM7QGVvNSK60T3OvWj3YfBRg==", + "dev": true, + "license": "MIT", + "peerDependencies": { + "@types/express": "*" + } + }, "node_modules/@types/cookiejar": { "version": "2.1.5", "resolved": "https://registry.npmjs.org/@types/cookiejar/-/cookiejar-2.1.5.tgz", @@ -23727,7 +23737,7 @@ }, "xstreamroll-sdk": { "name": "@stellar/streaming-sdk", - "version": "1.0.0", + "version": "1.1.0", "dependencies": { "@xstreamroll/types": "file:../packages/types" },