Repository: solid/community-server
Pull request #490 (2021-01-08 22:35:22)

Add transformSafely
Makes it easier for component implementers to keep streams safe (https://github.com/solid/community-server/issues/489).

The `transformSafely` function:
- catches errors on the source
- synchronous and asynchronous errors on transform
- synchronous and asynchronous errors on flush

Example usage:
- https://github.com/RubenVerborgh/solid-hue/commit/6221b9a56ab81c368cb53672e87a0e682dd1b6bf#diff-7f00ccca0a5d684cf6667c08205326e57b649644720382cc86c92276fdf0dca4R29-R45
- https://github.com/RubenVerborgh/solid-hue/commit/6221b9a56ab81c368cb53672e87a0e682dd1b6bf#diff-31541e15402fa0d5b58a80acc368dac7894e6f9fa9a1cbba2950eedfaa99c4b1R22-R45


Patch:
diff --git a/src/authorization/WebAclAuthorizer.ts b/src/authorization/WebAclAuthorizer.ts
index f464c428d..66565e2bd 100644
--- a/src/authorization/WebAclAuthorizer.ts
+++ b/src/authorization/WebAclAuthorizer.ts
@@ -64,7 +64,13 @@ export class WebAclAuthorizer extends Authorizer {
    */
   private checkPermission(agent: Credentials, store: Store, mode: string): void {
     const modeString = ACL[this.capitalize(mode) as 'Write' | 'Read' | 'Append' | 'Control'];
-    const auths = store.getQuads(null, ACL.mode, modeString, null).map((quad: Quad): Term => quad.subject);
+    const auths = this.getModePermissions(store, modeString);
+
+    // Having write permissions implies having append permissions
+    if (modeString === ACL.Append) {
+      auths.push(...this.getModePermissions(store, ACL.Write));
+    }
+
     if (!auths.some((term): boolean => this.hasAccess(agent, term, store))) {
       const isLoggedIn = typeof agent.webId === 'string';
       if (isLoggedIn) {
@@ -87,6 +93,15 @@ export class WebAclAuthorizer extends Authorizer {
     return `${mode[0].toUpperCase()}${mode.slice(1).toLowerCase()}`;
   }
 
+  /**
+   * Returns the identifiers of all authorizations that grant the given mode access for a resource.
+   * @param store - The store containing the quads of the acl resource.
+   * @param aclMode - A valid acl mode (ACL.Write/Read/...)
+   */
+  private getModePermissions(store: Store, aclMode: string): Term[] {
+    return store.getQuads(null, ACL.mode, aclMode, null).map((quad: Quad): Term => quad.subject);
+  }
+
   /**
    * Checks if the given agent has access to the modes specified by the given authorization.
    * @param agent - Credentials of agent that needs access.
diff --git a/src/util/StreamUtil.ts b/src/util/StreamUtil.ts
index 6e3ab02c9..dfcb0e9e7 100644
--- a/src/util/StreamUtil.ts
+++ b/src/util/StreamUtil.ts
@@ -1,5 +1,5 @@
-import type { Writable, ReadableOptions } from 'stream';
-import { Readable } from 'stream';
+import type { Writable, ReadableOptions, DuplexOptions } from 'stream';
+import { Readable, Transform } from 'stream';
 import arrayifyStream from 'arrayify-stream';
 import { getLoggerFor } from '../logging/LogUtil';
 import type { Guarded } from './GuardedStream';
@@ -43,6 +43,59 @@ export function pipeSafely<T extends Writable>(readable: NodeJS.ReadableStream,
   return guardStream(destination);
 }
 
+export interface AsyncTransformOptions<T = any> extends DuplexOptions {
+  /**
+   * Transforms data from the source by calling the `push` method
+   */
+  transform?: (this: Transform, data: T, encoding: string) => any | Promise<any>;
+
+  /**
+   * Performs any final actions after the source has ended
+   */
+  flush?: (this: Transform) => any | Promise<any>;
+}
+
+/**
+ * Transforms a stream, ensuring that all errors are forwarded.
+ * @param source - The stream to be transformed
+ * @param options - The transformation options
+ *
+ * @returns The transformed stream
+ */
+export function transformSafely<T = any>(
+  source: NodeJS.ReadableStream,
+  {
+    transform = function(data): void {
+      this.push(data);
+    },
+    flush = (): null => null,
+    ...options
+  }: AsyncTransformOptions<T> = {},
+):
+  Guarded<Transform> {
+  return pipeSafely(source, new Transform({
+    ...options,
+    async transform(data, encoding, callback): Promise<void> {
+      let error: Error | null = null;
+      try {
+        await transform.call(this, data, encoding);
+      } catch (err: unknown) {
+        error = err as Error;
+      }
+      callback(error);
+    },
+    async flush(callback): Promise<void> {
+      let error: Error | null = null;
+      try {
+        await flush.call(this);
+      } catch (err: unknown) {
+        error = err as Error;
+      }
+      callback(error);
+    },
+  }));
+}
+
 /**
  * Converts an iterable to a stream and applies an error guard so that it is {@link Guarded}.
  * @param iterable - Data to stream.


Test patch:
diff --git a/test/unit/authorization/WebAclAuthorizer.test.ts b/test/unit/authorization/WebAclAuthorizer.test.ts
index c3719d68a..79ee7460e 100644
--- a/test/unit/authorization/WebAclAuthorizer.test.ts
+++ b/test/unit/authorization/WebAclAuthorizer.test.ts
@@ -155,4 +155,20 @@ describe('A WebAclAuthorizer', (): void => {
     await expect(promise).rejects.toThrow('No ACL document found for root container');
     await expect(promise).rejects.toThrow(InternalServerError);
   });
+
+  it('allows an agent to append if they have write access.', async(): Promise<void> => {
+    credentials.webId = 'http://test.com/user';
+    identifier.path = 'http://test.com/foo';
+    permissions = {
+      read: false,
+      write: false,
+      append: true,
+    };
+    store.getRepresentation = async(): Promise<Representation> => ({ data: streamifyArray([
+      quad(nn('auth'), nn(`${acl}agent`), nn(credentials.webId!)),
+      quad(nn('auth'), nn(`${acl}accessTo`), nn(identifier.path)),
+      quad(nn('auth'), nn(`${acl}mode`), nn(`${acl}Write`)),
+    ]) } as Representation);
+    await expect(authorizer.handle({ identifier, permissions, credentials })).resolves.toBeUndefined();
+  });
 });
diff --git a/test/unit/util/StreamUtil.test.ts b/test/unit/util/StreamUtil.test.ts
index 42beb5bc8..f98d71f15 100644
--- a/test/unit/util/StreamUtil.test.ts
+++ b/test/unit/util/StreamUtil.test.ts
@@ -1,6 +1,7 @@
 import { PassThrough } from 'stream';
+import arrayifyStream from 'arrayify-stream';
 import streamifyArray from 'streamify-array';
-import { guardedStreamFrom, pipeSafely, readableToString } from '../../../src/util/StreamUtil';
+import { guardedStreamFrom, pipeSafely, transformSafely, readableToString } from '../../../src/util/StreamUtil';
 
 describe('StreamUtil', (): void => {
   describe('#readableToString', (): void => {
@@ -41,6 +42,104 @@ describe('StreamUtil', (): void => {
     });
   });
 
+  describe('#transformSafely', (): void => {
+    it('can transform a stream without arguments.', async(): Promise<void> => {
+      const source = streamifyArray([ 'data' ]);
+      const transformed = transformSafely(source);
+      transformed.setEncoding('utf8');
+      const result = await arrayifyStream(transformed);
+      expect(result).toEqual([ 'data' ]);
+    });
+
+    it('can transform a stream synchronously.', async(): Promise<void> => {
+      const source = streamifyArray([ 'data' ]);
+      const transformed = transformSafely<string>(source, {
+        encoding: 'utf8',
+        transform(data: string): void {
+          this.push(`${data}1`);
+          this.push(`${data}2`);
+        },
+        flush(): void {
+          this.push(`data3`);
+        },
+      });
+      const result = await arrayifyStream(transformed);
+      expect(result).toEqual([ 'data1', 'data2', 'data3' ]);
+    });
+
+    it('can transform a stream asynchronously.', async(): Promise<void> => {
+      const source = streamifyArray([ 'data' ]);
+      const transformed = transformSafely<string>(source, {
+        encoding: 'utf8',
+        async transform(data: string): Promise<void> {
+          await new Promise((resolve): any => setImmediate(resolve));
+          this.push(`${data}1`);
+          this.push(`${data}2`);
+        },
+        async flush(): Promise<void> {
+          await new Promise((resolve): any => setImmediate(resolve));
+          this.push(`data3`);
+        },
+      });
+      const result = await arrayifyStream(transformed);
+      expect(result).toEqual([ 'data1', 'data2', 'data3' ]);
+    });
+
+    it('catches source errors.', async(): Promise<void> => {
+      const error = new Error('stream error');
+      const source = new PassThrough();
+      const transformed = transformSafely<string>(source);
+      source.emit('error', error);
+      await expect(arrayifyStream(transformed)).rejects.toThrow(error);
+    });
+
+    it('catches synchronous errors on transform.', async(): Promise<void> => {
+      const error = new Error('stream error');
+      const source = streamifyArray([ 'data' ]);
+      const transformed = transformSafely<string>(source, {
+        transform(): never {
+          throw error;
+        },
+      });
+      await expect(arrayifyStream(transformed)).rejects.toThrow(error);
+    });
+
+    it('catches synchronous errors on flush.', async(): Promise<void> => {
+      const error = new Error('stream error');
+      const source = streamifyArray([ 'data' ]);
+      const transformed = transformSafely<string>(source, {
+        async flush(): Promise<never> {
+          await new Promise((resolve): any => setImmediate(resolve));
+          throw error;
+        },
+      });
+      await expect(arrayifyStream(transformed)).rejects.toThrow(error);
+    });
+
+    it('catches asynchronous errors on transform.', async(): Promise<void> => {
+      const error = new Error('stream error');
+      const source = streamifyArray([ 'data' ]);
+      const transformed = transformSafely<string>(source, {
+        transform(): never {
+          throw error;
+        },
+      });
+      await expect(arrayifyStream(transformed)).rejects.toThrow(error);
+    });
+
+    it('catches asynchronous errors on flush.', async(): Promise<void> => {
+      const error = new Error('stream error');
+      const source = streamifyArray([ 'data' ]);
+      const transformed = transformSafely<string>(source, {
+        async flush(): Promise<never> {
+          await new Promise((resolve): any => setImmediate(resolve));
+          throw error;
+        },
+      });
+      await expect(arrayifyStream(transformed)).rejects.toThrow(error);
+    });
+  });
+
   describe('#guardedStreamFrom', (): void => {
     it('converts data to a guarded stream.', async(): Promise<void> => {
       const data = [ 'a', 'b' ];