Repository: solid/community-server (ts)

Issue:
PUTting text file over root container closes the connection
To reproduce:

```shell
echo 'test' > test.txt
curl -H "Content-Type: text/plain" http://localhost:3000/. --upload-file test.txt -v
```

Logs:

```
2021-03-02T14:52:11.941Z [StreamUtil] warn: Piped stream errored with Unexpected "test" on line 1.
```

Expected:
- plaintext is rejected

Pull request:
fix: Prevent HttpRequest from being closed when downstream stream fails
Closes #646.

This PR is mostly comments and tests and not that much code. The code change is that the `pipeSafely` function now no longer calls the `pump` library to pipe when the source stream is an `HttpRequest`. Reason being that the `pump` library destroys the source stream if the destination stream errors. We want this behaviour, but not for the incoming request since that also closes the outgoing response.

I actually expected this to be a problem when the `pump` library got added to handle piping streams, but everything seemed to work so no change was needed. But apparently this does happen in specific cases so it's probably a timing issue where some errors are handled fast enough by our server and some aren't.

Patch:
diff --git a/src/ldp/http/BasicRequestParser.ts b/src/ldp/http/BasicRequestParser.ts
index b24c8f80d..58ada209d 100644
--- a/src/ldp/http/BasicRequestParser.ts
+++ b/src/ldp/http/BasicRequestParser.ts
@@ -1,4 +1,5 @@
 import type { HttpRequest } from '../../server/HttpRequest';
+import { InternalServerError } from '../../util/errors/InternalServerError';
 import type { Operation } from '../operations/Operation';
 import type { BodyParser } from './BodyParser';
 import type { MetadataExtractor } from './metadata/MetadataExtractor';
@@ -34,7 +35,7 @@ export class BasicRequestParser extends RequestParser {
   public async handle(request: HttpRequest): Promise<Operation> {
     const { method } = request;
     if (!method) {
-      throw new Error('No method specified on the HTTP request');
+      throw new InternalServerError('No method specified on the HTTP request');
     }
     const target = await this.targetExtractor.handleSafe({ request });
     const preferences = await this.preferenceParser.handleSafe({ request });
diff --git a/src/ldp/http/OriginalUrlExtractor.ts b/src/ldp/http/OriginalUrlExtractor.ts
index facf4865a..5e689daf3 100644
--- a/src/ldp/http/OriginalUrlExtractor.ts
+++ b/src/ldp/http/OriginalUrlExtractor.ts
@@ -1,5 +1,7 @@
 import type { TLSSocket } from 'tls';
 import type { HttpRequest } from '../../server/HttpRequest';
+import { BadRequestHttpError } from '../../util/errors/BadRequestHttpError';
+import { InternalServerError } from '../../util/errors/InternalServerError';
 import { parseForwarded } from '../../util/HeaderUtil';
 import { toCanonicalUriPath } from '../../util/PathUtil';
 import type { ResourceIdentifier } from '../representation/ResourceIdentifier';
@@ -18,7 +20,7 @@ export class OriginalUrlExtractor extends TargetExtractor {
 
   public async handle({ request: { url, connection, headers }}: { request: HttpRequest }): Promise<ResourceIdentifier> {
     if (!url) {
-      throw new Error('Missing URL');
+      throw new InternalServerError('Missing URL');
     }
 
     // Extract host and protocol (possibly overridden by the Forwarded/X-Forwarded-* header)
@@ -36,10 +38,10 @@ export class OriginalUrlExtractor extends TargetExtractor {
 
     // Perform a sanity check on the host
     if (!host) {
-      throw new Error('Missing Host header');
+      throw new BadRequestHttpError('Missing Host header');
     }
     if (/[/\\*]/u.test(host)) {
-      throw new Error(`The request has an invalid Host header: ${host}`);
+      throw new BadRequestHttpError(`The request has an invalid Host header: ${host}`);
     }
 
     // URL object applies punycode encoding to domain
diff --git a/src/ldp/representation/RepresentationMetadata.ts b/src/ldp/representation/RepresentationMetadata.ts
index be82e4c7a..a3c495ef5 100644
--- a/src/ldp/representation/RepresentationMetadata.ts
+++ b/src/ldp/representation/RepresentationMetadata.ts
@@ -1,6 +1,7 @@
 import { DataFactory, Store } from 'n3';
 import type { BlankNode, Literal, NamedNode, Quad, Term } from 'rdf-js';
 import { getLoggerFor } from '../../logging/LogUtil';
+import { InternalServerError } from '../../util/errors/InternalServerError';
 import { toSubjectTerm, toObjectTerm, toCachedNamedNode, isTerm } from '../../util/TermUtil';
 import { CONTENT_TYPE, CONTENT_TYPE_TERM } from '../../util/Vocabularies';
 import type { ResourceIdentifier } from './ResourceIdentifier';
@@ -265,7 +266,9 @@ export class RepresentationMetadata {
     }
     if (terms.length > 1) {
       this.logger.error(`Multiple results for ${typeof predicate === 'string' ? predicate : predicate.value}`);
-      throw new Error(`Multiple results for ${typeof predicate === 'string' ? predicate : predicate.value}`);
+      throw new InternalServerError(
+        `Multiple results for ${typeof predicate === 'string' ? predicate : predicate.value}`,
+      );
     }
     return terms[0];
   }
diff --git a/src/server/HttpRequest.ts b/src/server/HttpRequest.ts
index 1ba1dc48b..a071f4d46 100644
--- a/src/server/HttpRequest.ts
+++ b/src/server/HttpRequest.ts
@@ -5,3 +5,10 @@ import type { Guarded } from '../util/GuardedStream';
  * An incoming HTTP request;
  */
 export type HttpRequest = Guarded<IncomingMessage>;
+
+/**
+ * Checks if the given stream is an HttpRequest.
+ */
+export function isHttpRequest(stream: any): stream is HttpRequest {
+  return typeof stream.socket === 'object' && typeof stream.url === 'string' && typeof stream.method === 'string';
+}
diff --git a/src/storage/accessors/InMemoryDataAccessor.ts b/src/storage/accessors/InMemoryDataAccessor.ts
index b885e27a9..e748646b8 100644
--- a/src/storage/accessors/InMemoryDataAccessor.ts
+++ b/src/storage/accessors/InMemoryDataAccessor.ts
@@ -3,6 +3,7 @@ import arrayifyStream from 'arrayify-stream';
 import type { NamedNode } from 'rdf-js';
 import { RepresentationMetadata } from '../../ldp/representation/RepresentationMetadata';
 import type { ResourceIdentifier } from '../../ldp/representation/ResourceIdentifier';
+import { InternalServerError } from '../../util/errors/InternalServerError';
 import { NotFoundHttpError } from '../../util/errors/NotFoundHttpError';
 import type { Guarded } from '../../util/GuardedStream';
 import type { IdentifierStrategy } from '../../util/identifiers/IdentifierStrategy';
@@ -121,7 +122,7 @@ export class InMemoryDataAccessor implements DataAccessor {
         throw new NotFoundHttpError();
       }
       if (this.isDataEntry(parent)) {
-        throw new Error('Invalid path.');
+        throw new InternalServerError('Invalid path.');
       }
     }
 
diff --git a/src/storage/conversion/ChainedConverter.ts b/src/storage/conversion/ChainedConverter.ts
index 83d9d05a8..fc6de1298 100644
--- a/src/storage/conversion/ChainedConverter.ts
+++ b/src/storage/conversion/ChainedConverter.ts
@@ -1,5 +1,6 @@
 import type { Representation } from '../../ldp/representation/Representation';
 import { getLoggerFor } from '../../logging/LogUtil';
+import { InternalServerError } from '../../util/errors/InternalServerError';
 import { matchesMediaType } from './ConversionUtil';
 import type { RepresentationConverterArgs } from './RepresentationConverter';
 import { TypedRepresentationConverter } from './TypedRepresentationConverter';
@@ -79,7 +80,7 @@ export class ChainedConverter extends TypedRepresentationConverter {
 
     if (bestMatch.weight === 0) {
       this.logger.warn(`No match found between ${leftKeys} and ${rightKeys}`);
-      throw new Error(`No match found between ${leftKeys} and ${rightKeys}`);
+      throw new InternalServerError(`No match found between ${leftKeys} and ${rightKeys}`);
     }
 
     this.logger.debug(`${bestMatch.type} is the best match between ${leftKeys} and ${rightKeys}`);
diff --git a/src/storage/mapping/BaseFileIdentifierMapper.ts b/src/storage/mapping/BaseFileIdentifierMapper.ts
index e950b336c..e438842f0 100644
--- a/src/storage/mapping/BaseFileIdentifierMapper.ts
+++ b/src/storage/mapping/BaseFileIdentifierMapper.ts
@@ -2,6 +2,7 @@ import type { ResourceIdentifier } from '../../ldp/representation/ResourceIdenti
 import { getLoggerFor } from '../../logging/LogUtil';
 import { APPLICATION_OCTET_STREAM } from '../../util/ContentTypes';
 import { BadRequestHttpError } from '../../util/errors/BadRequestHttpError';
+import { InternalServerError } from '../../util/errors/InternalServerError';
 import { NotFoundHttpError } from '../../util/errors/NotFoundHttpError';
 import {
   decodeUriPathComponents,
@@ -98,7 +99,7 @@ export class BaseFileIdentifierMapper implements FileIdentifierMapper {
   public async mapFilePathToUrl(filePath: string, isContainer: boolean): Promise<ResourceLink> {
     if (!filePath.startsWith(this.rootFilepath)) {
       this.logger.error(`Trying to access file ${filePath} outside of ${this.rootFilepath}`);
-      throw new Error(`File ${filePath} is not part of the file storage at ${this.rootFilepath}`);
+      throw new InternalServerError(`File ${filePath} is not part of the file storage at ${this.rootFilepath}`);
     }
     const relative = filePath.slice(this.rootFilepath.length);
     let url: string;
diff --git a/src/util/StreamUtil.ts b/src/util/StreamUtil.ts
index 2eb536e70..a0a18030e 100644
--- a/src/util/StreamUtil.ts
+++ b/src/util/StreamUtil.ts
@@ -3,6 +3,7 @@ import { Readable, Transform } from 'stream';
 import arrayifyStream from 'arrayify-stream';
 import pump from 'pump';
 import { getLoggerFor } from '../logging/LogUtil';
+import { isHttpRequest } from '../server/HttpRequest';
 import type { Guarded } from './GuardedStream';
 import { guardStream } from './GuardedStream';
 
@@ -30,14 +31,30 @@ export async function readableToString(stream: Readable): Promise<string> {
  */
 export function pipeSafely<T extends Writable>(readable: NodeJS.ReadableStream, destination: T,
   mapError?: (error: Error) => Error): Guarded<T> {
-  // In case the input readable is guarded, it will no longer log errors since `pump` attaches a new error listener
-  pump(readable, destination, (error): void => {
-    if (error) {
-      logger.warn(`Piped stream errored with ${error.message}`);
-      // Make sure the final error can be handled in a normal streaming fashion
-      destination.emit('error', mapError ? mapError(error) : error);
-    }
-  });
+  // We never want to closes the incoming HttpRequest if there is an error
+  // since that also closes the outgoing HttpResponse.
+  // Since `pump` sends stream errors both up and down the pipe chain,
+  // in this case we need to make sure the error only goes down the chain.
+  if (isHttpRequest(readable)) {
+    readable.pipe(destination);
+    readable.on('error', (error): void => {
+      logger.warn(`HttpRequest errored with ${error.message}`);
+      // From https://nodejs.org/api/stream.html#stream_readable_pipe_destination_options :
+      // One important caveat is that if the Readable stream emits an error during processing,
+      // the Writable destination is not closed automatically. If an error occurs,
+      // it will be necessary to manually close each stream in order to prevent memory leaks.
+      destination.destroy(mapError ? mapError(error) : error);
+    });
+  } else {
+    // In case the input readable is guarded, it will no longer log errors since `pump` attaches a new error listener
+    pump(readable, destination, (error): void => {
+      if (error) {
+        logger.warn(`Piped stream errored with ${error.message}`);
+        // Make sure the final error can be handled in a normal streaming fashion
+        destination.emit('error', mapError ? mapError(error) : error);
+      }
+    });
+  }
   // Guarding the stream now means the internal error listeners of pump will be ignored
   // when checking if there is a valid error listener.
   return guardStream(destination);
diff --git a/src/util/handlers/WaterfallHandler.ts b/src/util/handlers/WaterfallHandler.ts
index 50a5dc5d0..4e7880369 100644
--- a/src/util/handlers/WaterfallHandler.ts
+++ b/src/util/handlers/WaterfallHandler.ts
@@ -48,7 +48,7 @@ export class WaterfallHandler<TIn, TOut> implements AsyncHandler<TIn, TOut> {
       handler = await this.findHandler(input);
     } catch {
       this.logger.warn('All handlers failed. This might be the consequence of calling handle before canHandle.');
-      throw new Error('All handlers failed');
+      throw new InternalServerError('All handlers failed');
     }
 
     return handler.handle(input);


Test patch:
diff --git a/test/integration/ServerFetch.test.ts b/test/integration/ServerFetch.test.ts
index e492468e6..3daa07b0a 100644
--- a/test/integration/ServerFetch.test.ts
+++ b/test/integration/ServerFetch.test.ts
@@ -68,6 +68,18 @@ describe('A Solid server', (): void => {
     expect(res.status).toBe(205);
   });
 
+  it('can handle PUT errors.', async(): Promise<void> => {
+    // There was a specific case where the following request caused the connection to close instead of error
+    const res = await fetch(baseUrl, {
+      method: 'PUT',
+      headers: {
+        'content-type': 'text/plain',
+      },
+      body: '"test"',
+    });
+    expect(res.status).toBe(400);
+  });
+
   it('can POST to create a container.', async(): Promise<void> => {
     const res = await fetch(baseUrl, {
       method: 'POST',
diff --git a/test/unit/server/HttpRequest.test.ts b/test/unit/server/HttpRequest.test.ts
new file mode 100644
index 000000000..461bcce85
--- /dev/null
+++ b/test/unit/server/HttpRequest.test.ts
@@ -0,0 +1,10 @@
+import { isHttpRequest } from '../../../src/server/HttpRequest';
+
+describe('HttpRequest', (): void => {
+  describe('#isHttpRequest', (): void => {
+    it('can identify HttpRequests.', async(): Promise<void> => {
+      expect(isHttpRequest({})).toBe(false);
+      expect(isHttpRequest({ socket: {}, method: 'GET', url: '/url' })).toBe(true);
+    });
+  });
+});
diff --git a/test/unit/util/StreamUtil.test.ts b/test/unit/util/StreamUtil.test.ts
index 70bfeb448..be3e0b1d6 100644
--- a/test/unit/util/StreamUtil.test.ts
+++ b/test/unit/util/StreamUtil.test.ts
@@ -1,8 +1,13 @@
 import { PassThrough } from 'stream';
 import arrayifyStream from 'arrayify-stream';
 import streamifyArray from 'streamify-array';
+import { isHttpRequest } from '../../../src/server/HttpRequest';
 import { guardedStreamFrom, pipeSafely, transformSafely, readableToString } from '../../../src/util/StreamUtil';
 
+jest.mock('../../../src/server/HttpRequest', (): any => ({
+  isHttpRequest: jest.fn(),
+}));
+
 describe('StreamUtil', (): void => {
   describe('#readableToString', (): void => {
     it('concatenates all elements of a Readable.', async(): Promise<void> => {
@@ -12,6 +17,10 @@ describe('StreamUtil', (): void => {
   });
 
   describe('#pipeSafely', (): void => {
+    beforeEach(async(): Promise<void> => {
+      (isHttpRequest as unknown as jest.Mock).mockClear();
+    });
+
     it('pipes data from one stream to the other.', async(): Promise<void> => {
       const input = streamifyArray([ 'data' ]);
       const output = new PassThrough();
@@ -56,6 +65,47 @@ describe('StreamUtil', (): void => {
       await new Promise(setImmediate);
       expect(input.destroyed).toBe(true);
     });
+
+    it('does not destroy the source stream if it is an HttpRequest.', async(): Promise<void> => {
+      (isHttpRequest as unknown as jest.Mock).mockReturnValueOnce(true);
+      const input = new PassThrough();
+      const output = new PassThrough();
+      const piped = pipeSafely(input, output);
+
+      // Catch errors to prevent problems in test output
+      output.on('error', (): void => {
+        // Empty
+      });
+
+      piped.destroy(new Error('error!'));
+      // Allow events to propagate
+      await new Promise(setImmediate);
+      expect(input.destroyed).toBe(false);
+    });
+
+    it('still sends errors downstream if the input is an HttpRequest.', async(): Promise<void> => {
+      (isHttpRequest as unknown as jest.Mock).mockReturnValueOnce(true);
+      const input = new PassThrough();
+      input.read = (): any => {
+        input.emit('error', new Error('error'));
+        return null;
+      };
+      const output = new PassThrough();
+      const piped = pipeSafely(input, output);
+      await expect(readableToString(piped)).rejects.toThrow('error');
+    });
+
+    it('can map errors if the input is an HttpRequest.', async(): Promise<void> => {
+      (isHttpRequest as unknown as jest.Mock).mockReturnValueOnce(true);
+      const input = streamifyArray([ 'data' ]);
+      input.read = (): any => {
+        input.emit('error', new Error('error'));
+        return null;
+      };
+      const output = new PassThrough();
+      const piped = pipeSafely(input, output, (): any => new Error('other error'));
+      await expect(readableToString(piped)).rejects.toThrow('other error');
+    });
   });
 
   describe('#transformSafely', (): void => {
@@ -158,8 +208,8 @@ describe('StreamUtil', (): void => {
 
   describe('#guardedStreamFrom', (): void => {
     it('converts data to a guarded stream.', async(): Promise<void> => {
-      const data = [ 'a', 'b' ];
-      await expect(readableToString(guardedStreamFrom(data))).resolves.toBe('ab');
+      await expect(readableToString(guardedStreamFrom([ 'a', 'b' ]))).resolves.toBe('ab');
+      await expect(readableToString(guardedStreamFrom('ab'))).resolves.toBe('ab');
     });
   });
 });