BackendSrv: incorporates validatePath in existing RxJs pipe (#107672)

* backendSrv: incorporates validatePath in existing RxJs pipe

* chore: updates after PR feedback

.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"],

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<FetchResponse<Uint8Array<ArrayBufferLike> | undefined>>;
+    options: BackendSrvRequest;
+    requestId: string;
+  }): Partial<Observer<Response>> {
+    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<T>(options: BackendSrvRequest): Observable<FetchResponse<T>> {
     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<T>(options).pipe(
+    return parseUrlFromOptions(options).pipe(
+      this.getFromFetchStream<T>(options),
       this.handleStreamResponse<T>(options),
       this.handleStreamError(options),
       this.handleStreamCancellation(options)
@@ -302,38 +325,36 @@ export class BackendSrv implements BackendService {
     return options;
   }
 
-  private getFromFetchStream<T>(options: BackendSrvRequest): Observable<FetchResponse<T>> {
-    const init = parseInitFromOptions(options);
+  private getFromFetchStream<T>(options: BackendSrvRequest): OperatorFunction<string, FetchResponse<T>> {
+    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<T>(response, responseType);
-        const fetchResponse: FetchResponse<T> = {
-          status,
-          statusText,
-          ok,
-          data,
-          headers,
-          url,
-          type,
-          redirected,
-          config: options,
-          traceId: response.headers.get(GRAFANA_TRACEID_HEADER) ?? undefined,
-        };
-        return fetchResponse;
-      })
-    );
+              const data = await parseResponseBody<T>(response, responseType);
+              const fetchResponse: FetchResponse<T> = {
+                status,
+                statusText,
+                ok,
+                data,
+                headers,
+                url,
+                type,
+                redirected,
+                config: options,
+                traceId: response.headers.get(GRAFANA_TRACEID_HEADER) ?? undefined,
+              };
+              return fetchResponse;
+            })
+          );
+        })
+      );
   }
 
   showApplicationErrorAlert(err: FetchError) {}

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

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');
+      }
     );
   });
 });

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<T>(
   return textData as T;
 }
 
-function serializeParams(data: Record<string, any>): string {
+function serializeParams(data: Record<string, string | number | boolean | Array<string | number | boolean>>): string {
   return Object.keys(data)
     .map((key) => {
       const value = data[key];
@@ -167,16 +168,22 @@ function serializeParams(data: Record<string, any>): 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<string> => {
+  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 => {