From a54453b94070c4450455f146cb569c047cf828c4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Hugo=20H=C3=A4ggmark?= Date: Tue, 19 Aug 2025 07:07:49 +0200 Subject: [PATCH] BackendSrv: incorporates validatePath in existing RxJs pipe (#107672) * backendSrv: incorporates validatePath in existing RxJs pipe * chore: updates after PR feedback --- .betterer.results | 3 +- public/app/core/services/backend_srv.ts | 207 ++++++++-------- public/app/core/specs/backend_srv.test.ts | 274 +++++++++++++++++++++- public/app/core/utils/fetch.test.ts | 25 +- public/app/core/utils/fetch.ts | 23 +- 5 files changed, 408 insertions(+), 124 deletions(-) diff --git a/.betterer.results b/.betterer.results index 695051344fb..6aec09ff9c7 100644 --- a/.betterer.results +++ b/.betterer.results @@ -1063,8 +1063,7 @@ exports[`better eslint`] = { [0, 0, 0, "Do not use any type assertions.", "1"], [0, 0, 0, "Do not use any type assertions.", "2"], [0, 0, 0, "Do not use any type assertions.", "3"], - [0, 0, 0, "Do not use any type assertions.", "4"], - [0, 0, 0, "Unexpected any. Specify a different type.", "5"] + [0, 0, 0, "Do not use any type assertions.", "4"] ], "public/app/core/utils/object.ts:5381": [ [0, 0, 0, "Do not use any type assertions.", "0"], diff --git a/public/app/core/services/backend_srv.ts b/public/app/core/services/backend_srv.ts index f518dc83df7..193a03a6847 100644 --- a/public/app/core/services/backend_srv.ts +++ b/public/app/core/services/backend_srv.ts @@ -1,5 +1,16 @@ import FingerprintJS from '@fingerprintjs/fingerprintjs'; -import { from, lastValueFrom, MonoTypeOperatorFunction, Observable, Subject, Subscription, throwError } from 'rxjs'; +import { + from, + lastValueFrom, + MonoTypeOperatorFunction, + Observable, + Observer, + OperatorFunction, + Subject, + Subscriber, + Subscription, + throwError, +} from 'rxjs'; import { fromFetch } from 'rxjs/fetch'; import { catchError, @@ -149,13 +160,6 @@ export class BackendSrv implements BackendService { const requestId = options.requestId ?? `chunked-${this.chunkRequestId++}`; const controller = new AbortController(); - let url: string; - try { - url = parseUrlFromOptions(options); - } catch (error) { - return throwError(() => error); - } - const init = parseInitFromOptions({ ...options, requestId, @@ -163,62 +167,10 @@ export class BackendSrv implements BackendService { }); return new Observable((observer) => { - let done = false; // Calling fromFetch explicitly avoids the request queue - const sub = this.dependencies.fromFetch(url, init).subscribe({ - next: (response) => { - const rsp = { - status: response.status, - statusText: response.statusText, - ok: response.ok, - headers: response.headers, - url: response.url, - type: response.type, - redirected: response.redirected, - config: options, - traceId: response.headers.get(GRAFANA_TRACEID_HEADER) ?? undefined, - data: undefined, - }; - - if (!response.body) { - observer.next(rsp); - observer.complete(); - return; - } - - const reader = response.body.getReader(); - async function process() { - while (reader && !done) { - if (controller.signal.aborted) { - reader.cancel(controller.signal.reason); - console.log(requestId, 'signal.aborted'); - return; - } - const chunk = await reader.read(); - observer.next({ - ...rsp, - data: chunk.value, - }); - if (chunk.done) { - done = true; - console.log(requestId, 'done'); - } - } - } - process() - .then(() => { - console.log(requestId, 'complete'); - observer.complete(); - }) // runs in background - .catch((e) => { - console.log(requestId, 'catch', e); - observer.error(e); - }); // from abort - }, - error: (e) => { - observer.error(e); - }, - }); + const sub = parseUrlFromOptions(options) + .pipe(mergeMap((url) => this.dependencies.fromFetch(url, init))) + .subscribe(this.getChunkedResponseObserver({ controller, observer, options, requestId })); return function unsubscribe() { console.log(requestId, 'unsubscribe'); @@ -228,6 +180,76 @@ export class BackendSrv implements BackendService { }); } + private getChunkedResponseObserver({ + controller, + observer, + options, + requestId, + }: { + controller: AbortController; + observer: Subscriber | undefined>>; + options: BackendSrvRequest; + requestId: string; + }): Partial> { + let done = false; + return { + next: (response) => { + const rsp = { + status: response.status, + statusText: response.statusText, + ok: response.ok, + headers: response.headers, + url: response.url, + type: response.type, + redirected: response.redirected, + config: options, + traceId: response.headers.get(GRAFANA_TRACEID_HEADER) ?? undefined, + data: undefined, + }; + + if (!response.body) { + observer.next(rsp); + observer.complete(); + return; + } + + const reader = response.body.getReader(); + + // Setup onabort callback so that we can cancel the reader properly + controller.signal.onabort = () => { + reader.cancel(controller.signal.reason); + console.log(requestId, 'signal.aborted'); + }; + + async function process() { + while (reader && !done) { + const chunk = await reader.read(); + observer.next({ + ...rsp, + data: chunk.value, + }); + if (chunk.done) { + done = true; + console.log(requestId, 'done'); + } + } + } + process() + .then(() => { + console.log(requestId, 'complete'); + observer.complete(); + }) // runs in background + .catch((e) => { + console.log(requestId, 'catch', e); + observer.error(e); + }); // from abort + }, + error: (e) => { + observer.error(e); + }, + }; + } + private internalFetch(options: BackendSrvRequest): Observable> { if (options.requestId) { this.inFlightRequests.next(options.requestId); @@ -248,7 +270,8 @@ export class BackendSrv implements BackendService { options.headers['X-Grafana-Device-Id'] = `${this.deviceID}`; } - return this.getFromFetchStream(options).pipe( + return parseUrlFromOptions(options).pipe( + this.getFromFetchStream(options), this.handleStreamResponse(options), this.handleStreamError(options), this.handleStreamCancellation(options) @@ -302,38 +325,36 @@ export class BackendSrv implements BackendService { return options; } - private getFromFetchStream(options: BackendSrvRequest): Observable> { - const init = parseInitFromOptions(options); + private getFromFetchStream(options: BackendSrvRequest): OperatorFunction> { + return (inputStream) => + inputStream.pipe( + mergeMap((url) => { + const init = parseInitFromOptions(options); - let url: string; - try { - url = parseUrlFromOptions(options); - } catch (error) { - return throwError(() => error); - } + return this.dependencies.fromFetch(url, init).pipe( + mergeMap(async (response) => { + const { status, statusText, ok, headers, url, type, redirected } = response; - return this.dependencies.fromFetch(url, init).pipe( - mergeMap(async (response) => { - const { status, statusText, ok, headers, url, type, redirected } = response; + const responseType = options.responseType ?? (isContentTypeJson(headers) ? 'json' : undefined); - const responseType = options.responseType ?? (isContentTypeJson(headers) ? 'json' : undefined); - - const data = await parseResponseBody(response, responseType); - const fetchResponse: FetchResponse = { - status, - statusText, - ok, - data, - headers, - url, - type, - redirected, - config: options, - traceId: response.headers.get(GRAFANA_TRACEID_HEADER) ?? undefined, - }; - return fetchResponse; - }) - ); + const data = await parseResponseBody(response, responseType); + const fetchResponse: FetchResponse = { + status, + statusText, + ok, + data, + headers, + url, + type, + redirected, + config: options, + traceId: response.headers.get(GRAFANA_TRACEID_HEADER) ?? undefined, + }; + return fetchResponse; + }) + ); + }) + ); } showApplicationErrorAlert(err: FetchError) {} diff --git a/public/app/core/specs/backend_srv.test.ts b/public/app/core/specs/backend_srv.test.ts index 097788913d8..9ad572dbac3 100644 --- a/public/app/core/specs/backend_srv.test.ts +++ b/public/app/core/specs/backend_srv.test.ts @@ -1,4 +1,4 @@ -import { Observable, of, lastValueFrom } from 'rxjs'; +import { Observable, of, lastValueFrom, throwError } from 'rxjs'; import { fromFetch } from 'rxjs/fetch'; import { delay } from 'rxjs/operators'; @@ -782,22 +782,19 @@ describe('backendSrv', () => { await expect(promise).rejects.toThrow(PathValidationError); }); - it('should sanitise paths when calling .fetch', (done) => { + it('should sanitise paths when calling .fetch', async () => { const { backendSrv } = getTestContext(); const maliciousUrl = '/api/users/%2e%2e/admin'; - const observable = backendSrv.fetch({ url: maliciousUrl, method: 'GET', validatePath: true }); + await expect(backendSrv.fetch({ url: maliciousUrl, method: 'GET', validatePath: true })).toEmitValuesWith( + (received) => { + expect(received.length).toEqual(1); - observable.subscribe({ - next: () => { - throw new Error('Should not succeed'); - }, - error: (err) => { - expect(err).toBeInstanceOf(PathValidationError); - expect(err.message).toBe('Invalid request path'); - done(); - }, - }); + const processed: FetchError = received[0]; + expect(processed).toBeInstanceOf(PathValidationError); + expect(processed.message).toBe('Invalid request path'); + } + ); }); }); @@ -883,4 +880,255 @@ describe('backendSrv', () => { }); }); }); + + describe('chunked', () => { + beforeEach(() => { + // we do a bunch of console.log in the chunked function + jest.spyOn(console, 'log').mockImplementation(() => {}); + }); + + afterEach(() => { + jest.resetAllMocks(); + }); + + describe('when making a successful chunked request', () => { + it('then it should return chunks of data', async () => { + const url = '/api/chunked-data'; + const { backendSrv, fromFetchMock } = getTestContext({ url }); + + // Mock a ReadableStream with chunks + const chunks = [new Uint8Array([1, 2, 3]), new Uint8Array([4, 5, 6]), new Uint8Array([7, 8, 9])]; + + let chunkIndex = 0; + const mockReader = { + read: jest.fn().mockImplementation(() => { + if (chunkIndex < chunks.length) { + return Promise.resolve({ + done: false, + value: chunks[chunkIndex++], + }); + } + return Promise.resolve({ done: true, value: undefined }); + }), + cancel: jest.fn(), + }; + + const mockBody = { + getReader: jest.fn().mockReturnValue(mockReader), + }; + + const mockResponse = { + ok: true, + status: 200, + statusText: 'OK', + headers: [], + body: mockBody, + url, + type: 'basic', + redirected: false, + }; + + fromFetchMock.mockReturnValue(of(mockResponse)); + + const options = { url, method: 'GET' }; + + await expect(backendSrv.chunked(options)).toEmitValuesWith((received) => { + const { body, ...rest } = mockResponse; + const config = { ...options }; + const expected = { ...rest, config }; + expect(received).toHaveLength(4); // 3 chunks + 1 final response + expect(received[0]).toEqual({ ...expected, data: new Uint8Array([1, 2, 3]) }); + expect(received[1]).toEqual({ ...expected, data: new Uint8Array([4, 5, 6]) }); + expect(received[2]).toEqual({ ...expected, data: new Uint8Array([7, 8, 9]) }); + expect(received[3]).toEqual({ ...expected, data: undefined }); + expect(mockReader.read).toHaveBeenCalledTimes(4); + }); + }); + }); + + describe('when request is cancelled', () => { + it('then it should abort the request and cancel the reader', async () => { + jest.useFakeTimers(); + const url = '/api/chunked-data'; + const { backendSrv, fromFetchMock } = getTestContext({ url }); + + const mockReader = { + read: jest.fn().mockImplementation(() => { + return new Promise(() => {}); // Never resolves + }), + cancel: jest.fn(), + }; + + const mockBody = { + getReader: jest.fn().mockReturnValue(mockReader), + }; + + const mockResponse = { + ok: true, + status: 200, + statusText: 'OK', + headers: [], + body: mockBody, + url, + type: 'basic', + redirected: false, + }; + + fromFetchMock.mockReturnValue(of(mockResponse)); + + const options = { url, method: 'GET' }; + + const subscription = backendSrv.chunked(options).subscribe(); + + // Cancel the request + subscription.unsubscribe(); + + // Fast-forward until all timers have been executed + jest.advanceTimersByTime(100); + + expect(mockReader.cancel).toHaveBeenCalled(); + }); + }); + + describe('when request throws an error', () => { + it('then it should complete immediately', async () => { + const url = '/api/chunked-data'; + const { backendSrv, fromFetchMock } = getTestContext({ url }); + + fromFetchMock.mockReturnValue(throwError(() => new Error('Server error'))); + + const options = { url, method: 'GET' }; + + await expect(backendSrv.chunked(options)).toEmitValuesWith((received) => { + expect(received).toHaveLength(1); + + const error: FetchError = received[0]; + + expect(error).toBeInstanceOf(Error); + expect(error.message).toEqual('Server error'); + }); + }); + }); + + describe('when response has no body', () => { + it('then it should complete immediately', async () => { + const url = '/api/chunked-data'; + const { backendSrv, fromFetchMock } = getTestContext({ url }); + + const mockResponse = { + ok: true, + status: 200, + statusText: 'OK', + headers: [], + body: null, + url, + type: 'basic', + redirected: false, + }; + + fromFetchMock.mockReturnValue(of(mockResponse)); + + const options = { url, method: 'GET' }; + + await expect(backendSrv.chunked(options)).toEmitValuesWith((received) => { + const { body, ...rest } = mockResponse; + const config = { ...options }; + const expected = { ...rest, config }; + expect(received).toHaveLength(1); + expect(received[0]).toEqual({ ...expected, data: undefined }); + }); + }); + }); + + describe('when validatePath is true and url is malicious', () => { + it('then it should throw an PathValidationError', async () => { + const url = '/api/users/%2e%2e/admin'; + const { backendSrv, fromFetchMock } = getTestContext({ url }); + + const mockResponse = { + ok: true, + status: 200, + statusText: 'OK', + headers: [], + body: null, + url, + type: 'basic', + redirected: false, + }; + + fromFetchMock.mockReturnValue(of(mockResponse)); + + const options = { url, method: 'GET', validatePath: true }; + + await expect(backendSrv.chunked(options)).toEmitValuesWith((received) => { + expect(received).toHaveLength(1); + + const error: FetchError = received[0]; + + expect(error).toBeInstanceOf(PathValidationError); + expect(error.message).toEqual('Invalid request path'); + }); + }); + }); + + describe('when validatePath is true and url is not malicious', () => { + it('then it should return correct chunks', async () => { + const url = '/api/chunked-data'; + const { backendSrv, fromFetchMock } = getTestContext({ url }); + + const mockResponse = { + ok: true, + status: 200, + statusText: 'OK', + headers: [], + body: null, + url, + type: 'basic', + redirected: false, + }; + + fromFetchMock.mockReturnValue(of(mockResponse)); + + const options = { url, method: 'GET', validatePath: true }; + + await expect(backendSrv.chunked(options)).toEmitValuesWith((received) => { + const { body, ...rest } = mockResponse; + const config = { ...options }; + const expected = { ...rest, config }; + expect(received).toHaveLength(1); + expect(received[0]).toEqual({ ...expected, data: undefined }); + }); + }); + }); + + describe('when validatePath is false and url is malicious', () => { + it('then it should return correct chunks', async () => { + const url = '/api/users/%2e%2e/admin'; + const { backendSrv, fromFetchMock } = getTestContext({ url }); + + const mockResponse = { + ok: true, + status: 200, + statusText: 'OK', + headers: [], + body: null, + url, + type: 'basic', + redirected: false, + }; + + fromFetchMock.mockReturnValue(of(mockResponse)); + + const options = { url, method: 'GET', validatePath: false }; + + await expect(backendSrv.chunked(options)).toEmitValuesWith((received) => { + const { body, ...rest } = mockResponse; + const config = { ...options }; + const expected = { ...rest, config }; + expect(received).toHaveLength(1); + expect(received[0]).toEqual({ ...expected, data: undefined }); + }); + }); + }); + }); }); diff --git a/public/app/core/utils/fetch.test.ts b/public/app/core/utils/fetch.test.ts index cd613132413..b764f61bb49 100644 --- a/public/app/core/utils/fetch.test.ts +++ b/public/app/core/utils/fetch.test.ts @@ -27,20 +27,29 @@ describe('parseUrlFromOptions', () => { ${{ id: [] }} | ${'api/dashboard'} | ${'api/dashboard'} `( "when called with params: '$params' and url: '$url' then result should be '$expected'", - ({ params, url, expected }) => { - expect(parseUrlFromOptions({ params, url })).toEqual(expected); + async ({ params, url, expected }) => { + await expect(parseUrlFromOptions({ params, url })).toEmitValuesWith((received) => { + expect(received).toHaveLength(1); + expect(received[0]).toEqual(expected); + }); } ); - it('should validate the path if validatePath is true', () => { - expect(() => parseUrlFromOptions({ url: '/api/users/%2e%2e/admin', validatePath: true })).toThrow( - PathValidationError + it('should validate the path if validatePath is true', async () => { + await expect(parseUrlFromOptions({ url: '/api/users/%2e%2e/admin', validatePath: true })).toEmitValuesWith( + (received) => { + expect(received).toHaveLength(1); + expect(received[0]).toBeInstanceOf(PathValidationError); + } ); }); - it('should not validate the path if validatePath is false', () => { - expect(parseUrlFromOptions({ url: '/api/users/%2e%2e/admin', validatePath: false })).toEqual( - '/api/users/%2e%2e/admin' + it('should not validate the path if validatePath is false', async () => { + await expect(parseUrlFromOptions({ url: '/api/users/%2e%2e/admin', validatePath: false })).toEmitValuesWith( + (received) => { + expect(received).toHaveLength(1); + expect(received[0]).toEqual('/api/users/%2e%2e/admin'); + } ); }); }); diff --git a/public/app/core/utils/fetch.ts b/public/app/core/utils/fetch.ts index a456f35e066..e714e944dc9 100644 --- a/public/app/core/utils/fetch.ts +++ b/public/app/core/utils/fetch.ts @@ -1,4 +1,5 @@ import { omitBy } from 'lodash'; +import { Observable, of, throwError } from 'rxjs'; import { deprecationWarning, validatePath } from '@grafana/data'; import { BackendSrvRequest } from '@grafana/runtime'; @@ -152,7 +153,7 @@ export async function parseResponseBody( return textData as T; } -function serializeParams(data: Record): string { +function serializeParams(data: Record>): string { return Object.keys(data) .map((key) => { const value = data[key]; @@ -167,16 +168,22 @@ function serializeParams(data: Record): string { /** * Formats and validates the URL. * If options.validatePath is true, this will throw an exception if the URL fails validation. + * @param options - The options to parse. + * @returns An observable that emits the parsed URL or an error if the URL fails validation. */ -export const parseUrlFromOptions = (options: BackendSrvRequest): string => { - const cleanParams = omitBy(options.params, (v) => v === undefined || (v && v.length === 0)); - const serializedParams = serializeParams(cleanParams); +export const parseUrlFromOptions = (options: BackendSrvRequest): Observable => { + try { + const cleanParams = omitBy(options.params, (v) => v === undefined || (v && v.length === 0)); + const serializedParams = serializeParams(cleanParams); - const url = options.validatePath // - ? validatePath(options.url) - : options.url; + const url = options.validatePath // + ? validatePath(options.url) + : options.url; - return options.params && serializedParams.length ? `${url}?${serializedParams}` : url; + return options.params && serializedParams.length ? of(`${url}?${serializedParams}`) : of(url); + } catch (error) { + return throwError(() => error); + } }; export const parseCredentials = (options: BackendSrvRequest): RequestCredentials => {