Implement maxUploadFileSize

hub/config-schema.json

@@ -126,6 +126,12 @@
             },
             "type": "object"
         },
+        "maxFileUploadSize": {
+            "default": 20,
+            "description": "The maximum allowed POST body size in megabytes. \nThe content-size header is checked, and the POST body stream \nis monitoring while streaming from the client. \n[Recommended] Minimum 100KB (or approximately 0.1MB)",
+            "minimum": 0.1,
+            "type": "number"
+        },
         "pageSize": {
             "default": 100,
             "maximum": 4096,

hub/package.json

@@ -11,12 +11,12 @@
     "@google-cloud/storage": "^2.5.0",
     "ajv": "^6.10.0",
     "aws-sdk": "^2.482.0",
-    "bitcoinjs-lib": "^5.0.5",
+    "bitcoinjs-lib": "^5.1.2",
     "blockstack": "^19.2.2",
-    "body-parser": "^1.18.3",
+    "body-parser": "^1.19.0",
     "cors": "^2.8.5",
-    "express": "^4.16.4",
-    "express-winston": "^3.1.0",
+    "express": "^4.17.1",
+    "express-winston": "^3.2.1",
     "fs-extra": "^8.1.0",
     "jsontokens": "^2.0.2",
     "lru-cache": "^5.1.1",

hub/src/server/config.ts

@@ -58,6 +58,7 @@ export interface HubConfigInterface {
   bucket?: string;
   pageSize?: number;
   cacheControl?: string;
+  maxFileUploadSize?: number;
   argsTransport?: LoggingConfig;
   proofsConfig?: ProofCheckerConfigInterface;
   driver?: DriverName;
@@ -124,6 +125,14 @@ class HubConfig implements HubConfigInterface {
    */
   pageSize? = 100;
   cacheControl? = 'public, max-age=1';
+  /**
+   * The maximum allowed POST body size in megabytes. 
+   * The content-size header is checked, and the POST body stream 
+   * is monitoring while streaming from the client. 
+   * [Recommended] Minimum 100KB (or approximately 0.1MB)
+   * @minimum 0.1
+   */
+  maxFileUploadSize? = 20;
   /**
    * @minimum 0
    * @maximum 65535

hub/src/server/errors.ts

@@ -49,3 +49,17 @@ export class InvalidInputError extends Error {
     this.name = this.constructor.name
   }
 }
+
+export class ContentLengthHeaderRequiredError extends Error {
+  constructor (message: string) {
+    super(message)
+    this.name = this.constructor.name
+  }
+}
+
+export class PayloadTooLargeError extends Error {
+  constructor (message: string) {
+    super(message)
+    this.name = this.constructor.name
+  }
+}

hub/src/server/http.ts

@@ -68,13 +68,17 @@ export function makeHttpServer(config: HubConfigInterface): { app: express.Appli
         if (err instanceof errors.ValidationError) {
           writeResponse(res, { message: err.message, error: err.name }, 401)
         } else if (err instanceof errors.AuthTokenTimestampValidationError) {
-          writeResponse(res, { message: err.message, error: err.name  }, 401)
+          writeResponse(res, { message: err.message, error: err.name }, 401)
         } else if (err instanceof errors.BadPathError) {
-          writeResponse(res, { message: err.message, error: err.name  }, 403)
+          writeResponse(res, { message: err.message, error: err.name }, 403)
         } else if (err instanceof errors.NotEnoughProofError) {
-          writeResponse(res, { message: err.message, error: err.name  }, 402)
+          writeResponse(res, { message: err.message, error: err.name }, 402)
         } else if (err instanceof errors.ConflictError) {
-          writeResponse(res, { message: err.message, error: err.name  }, 409) 
+          writeResponse(res, { message: err.message, error: err.name }, 409) 
+        } else if (err instanceof errors.ContentLengthHeaderRequiredError) {
+          writeResponse(res, { message: err.message, error: err.name }, 411)
+        } else if (err instanceof errors.PayloadTooLargeError) {
+          writeResponse(res, { message: err.message, error: err.name }, 413)
         } else {
           writeResponse(res, { message: 'Server Error' }, 500)
         }
@@ -148,9 +152,9 @@ export function makeHttpServer(config: HubConfigInterface): { app: express.Appli
   })
 
   app.post(
-    /^\/list-files\/([a-zA-Z0-9]+)\/?/, express.json(),
+    /^\/list-files\/([a-zA-Z0-9]+)\/?/, express.json({ limit: 4096 }),
     (req: express.Request, res: express.Response) => {
-      // sanity check...
+      // sanity check... should never be reached if the express json parser is working correctly
       if (parseInt(req.headers['content-length']) > 4096) {
         writeResponse(res, { message: 'Invalid JSON: too long'}, 400)
         return
@@ -179,9 +183,9 @@ export function makeHttpServer(config: HubConfigInterface): { app: express.Appli
 
   app.post(
     /^\/revoke-all\/([a-zA-Z0-9]+)\/?/, 
-    express.json(),
+    express.json({ limit: 4096 }),
     (req: express.Request, res: express.Response) => {
-      // sanity check...
+      // sanity check... should never be reached if the express json parser is working correctly
       if (parseInt(req.headers['content-length']) > 4096) {
         writeResponse(res, { message: 'Invalid JSON: too long'}, 400)
         return
@@ -225,6 +229,7 @@ export function makeHttpServer(config: HubConfigInterface): { app: express.Appli
     const readURLPrefix = server.getReadURLPrefix()
     writeResponse(res, { 'challenge_text': challengeText,
                          'latest_auth_version': LATEST_AUTH_VERSION,
+                         'max_file_upload_size_megabytes': server.maxFileUploadSizeMB,
                          'read_url_prefix': readURLPrefix }, 200)
   })
 

hub/src/server/server.ts

@@ -1,14 +1,14 @@
 
 
 import { validateAuthorizationHeader, getAuthenticationScopes, AuthScopeValues } from './authentication'
-import { ValidationError, DoesNotExist } from './errors'
+import { ValidationError, DoesNotExist, ContentLengthHeaderRequiredError, PayloadTooLargeError } from './errors'
 import { ProofChecker } from './ProofChecker'
 import { AuthTimestampCache } from './revocations'
 
-import { Readable } from 'stream'
+import { Readable, PassThrough } from 'stream'
 import { DriverModel, PerformWriteArgs, PerformRenameArgs, PerformDeleteArgs, PerformListFilesArgs, ListFilesStatResult, ListFilesResult } from './driverModel'
 import { HubConfigInterface } from './config'
-import { logger, generateUniqueID } from './utils'
+import { logger, generateUniqueID, bytesToMegabytes, megabytesToBytes, pipelineAsync } from './utils'
 
 export class HubServer {
   driver: DriverModel
@@ -20,6 +20,8 @@ export class HubServer {
   validHubUrls?: Array<string>
   authTimestampCache: AuthTimestampCache
   config: HubConfigInterface
+  maxFileUploadSizeMB: number
+  maxFileUploadSizeBytes: number
 
   constructor(driver: DriverModel, proofChecker: ProofChecker, config: HubConfigInterface) {
     this.driver = driver
@@ -31,6 +33,9 @@ export class HubServer {
     this.readURL = config.readURL
     this.requireCorrectHubUrl = config.requireCorrectHubUrl || false
     this.authTimestampCache = new AuthTimestampCache(this.getReadURLPrefix(), driver, config.authTimestampCacheSize)
+    this.maxFileUploadSizeMB = (config.maxFileUploadSize || 20)
+    // megabytes to bytes
+    this.maxFileUploadSizeBytes = megabytesToBytes(this.maxFileUploadSizeMB)
   }
 
   async handleAuthBump(address: string, oldestValidTimestamp: number, requestHeaders: { authorization?: string }) {
@@ -208,14 +213,25 @@ export class HubServer {
       }
     }
 
-    const writeCommand: PerformWriteArgs = {
-      storageTopLevel: address,
-      path, stream, contentType,
-      contentLength: parseInt(requestHeaders['content-length'] as string)
+    await this.proofChecker.checkProofs(address, path, this.getReadURLPrefix())
+
+    const contentLengthHeader = requestHeaders['content-length'] as string
+    const contentLengthBytes = parseInt(contentLengthHeader)
+    const isLengthFinite = Number.isFinite(contentLengthBytes)
+
+    if (!isLengthFinite) {
+      const errMsg = `A valid 'Content-Length' header must be passed. Received header "${contentLengthHeader}"`
+      logger.warn(`${errMsg}, address: ${address}`)
+      throw new ContentLengthHeaderRequiredError(errMsg)
+    }
+
+    if (contentLengthBytes > this.maxFileUploadSizeBytes) {
+      const errMsg = `Max file upload size is ${this.maxFileUploadSizeMB} megabytes. ` + 
+        `Rejected Content-Length of ${bytesToMegabytes(contentLengthBytes, 4)} megabytes`
+      logger.warn(`${errMsg}, address: ${address}`)
+      throw new PayloadTooLargeError(errMsg)
     }
 
-    await this.proofChecker.checkProofs(address, path, this.getReadURLPrefix())
-
     if (isArchivalRestricted) {
       const historicalPath = this.getHistoricalFileName(path)
       try {
@@ -240,7 +256,45 @@ export class HubServer {
       }
     }
 
-    const readURL = await this.driver.performWrite(writeCommand)
+    // Use the stream pipe API to monitor a stream with correct backpressure handling. This 
+    // avoids buffering entire streams in memory and hooks up all the correct events for 
+    // cleanup and error handling. See https://nodejs.org/api/stream.html#stream_three_states
+
+    // Two stream listeners are required to enforce size limits: the original stream reader 
+    // implemented by drivers for uploading to storage, and another for monitoring upload size. 
+    // Stream listeners are greedy so two PassThrough streams are used, and the pipeline API
+    // is setup *before* reading so that both listeners receive all notifications. 
+    // See https://stackoverflow.com/a/51143558/794962
+
+    // Create a PassThrough stream to monitor streaming size. 
+    const monitorStream = new PassThrough()
+    const monitorPipeline = pipelineAsync(stream, monitorStream)
+
+    // Create a PassThrough stream to give to driver for uploading to storage backend. 
+    const uploadStream = new PassThrough()
+    const uploadPipeline = pipelineAsync(stream, uploadStream)
+
+    // Now that the PassThrough streams and pipes are setup, the data event listener
+    // for the size monitor can be setup without either stream listeners missing any chunks. 
+    let monitoredContentSize = 0
+    monitorStream.on('data', (chunk: Buffer) => {
+      monitoredContentSize += chunk.length
+      if (monitoredContentSize > this.maxFileUploadSizeBytes) {
+        const errMsg = `Max file upload size is ${this.maxFileUploadSizeMB} megabytes. ` + 
+          `Rejected POST body stream of ${bytesToMegabytes(monitoredContentSize, 4)} megabytes`
+        logger.warn(`${errMsg}, address: ${address}`)
+        const error = new PayloadTooLargeError(errMsg)
+        stream.destroy(error)
+        uploadStream.destroy(error)
+      }
+    })
+
+    const writeCommand: PerformWriteArgs = {
+      storageTopLevel: address,
+      path, stream: uploadStream, contentType,
+      contentLength: contentLengthBytes
+    }
+    const [readURL] = await Promise.all([this.driver.performWrite(writeCommand), uploadPipeline, monitorPipeline])
     const driverPrefix = this.driver.getReadURLPrefix()
     const readURLPrefix = this.getReadURLPrefix()
     if (readURLPrefix !== driverPrefix && readURL.startsWith(driverPrefix)) {

hub/src/server/utils.ts

@@ -43,6 +43,14 @@ export function getDriverClass(driver: DriverName): DriverConstructor & DriverSt
   }
 }
 
+export function megabytesToBytes(megabytes: number) {
+  return megabytes * 1024 * 1024
+}
+
+export function bytesToMegabytes(bytes: number, decimals = 2) {
+  return Number.parseFloat((bytes / 1024 / 1024).toFixed(decimals))
+}
+
 export function dateToUnixTimeSeconds(date: Date) {
   return Math.round(date.getTime() / 1000)
 }

hub/test/src/testHttp.ts

@@ -19,6 +19,8 @@ import InMemoryDriver from './testDrivers/InMemoryDriver'
 import { MockAuthTimestampCache } from './MockAuthTimestampCache'
 import { HubConfigInterface } from '../../src/server/config'
 import { PassThrough } from 'stream';
+import * as errors from '../../src/server/errors'
+import { timeout } from '../../src/server/utils';
 
 const TEST_SERVER_NAME = 'test-server'
 const TEST_AUTH_CACHE_SIZE = 10
@@ -126,7 +128,13 @@ export function testHttpWithInMemoryDriver() {
     const fetch = NodeFetch
     const inMemoryDriver = await InMemoryDriver.spawn()
     try {
-      const { app } = makeHttpServer({ driverInstance: inMemoryDriver, serverName: TEST_SERVER_NAME, authTimestampCacheSize: TEST_AUTH_CACHE_SIZE })
+      const { app, server } = makeHttpServer({ 
+        driverInstance: inMemoryDriver, 
+        serverName: TEST_SERVER_NAME, 
+        authTimestampCacheSize: TEST_AUTH_CACHE_SIZE,      
+        // ~52 byte max limit
+        maxFileUploadSize: 0.00005 
+      })
       const sk = testPairs[1]
       const fileContents = sk.toWIF()
       const blob = Buffer.from(fileContents)
@@ -180,7 +188,29 @@ export function testHttpWithInMemoryDriver() {
         .send(blob)
         .expect(409)
 
-      t.end()
+      await request(app).post(path)
+        .set('Content-Type', 'application/octet-stream')
+        .set('Authorization', authorization)
+        .set('Content-Length', '9999999')
+        .expect(413)
+
+      try {
+        const largePayload = new PassThrough()
+        largePayload.end('x'.repeat(1000))
+        await server.handleRequest(address, 'helloWorld', { 'content-type': 'application/octet-stream', 'content-length': 'nope', authorization: authorization}, largePayload);
+        t.fail('should have thrown content length header required')
+      } catch (err) {
+        t.throws(() => { throw err }, errors.ContentLengthHeaderRequiredError, 'should have thrown content length header required')
+      }
+
+      try {
+        const largePayload = new PassThrough()
+        largePayload.end('x'.repeat(1000))
+        await server.handleRequest(address, 'helloWorld', { 'content-type': 'application/octet-stream', 'content-length': '10', authorization: authorization}, largePayload);
+        t.fail('payload should have been detected as too large')
+      } catch (err) {
+        t.throws(() => { throw err }, Error, 'payload should have been detected as too large')
+      }
 
     } finally {
       inMemoryDriver.dispose()
@@ -190,7 +220,11 @@ export function testHttpWithInMemoryDriver() {
   test('handle revocation via POST', async (t) => {
     const inMemoryDriver = await InMemoryDriver.spawn()
     try {
-      const { app } = makeHttpServer({ driverInstance: inMemoryDriver, serverName: TEST_SERVER_NAME, authTimestampCacheSize: TEST_AUTH_CACHE_SIZE })
+      const { app } = makeHttpServer({ 
+        driverInstance: inMemoryDriver, 
+        serverName: TEST_SERVER_NAME, 
+        authTimestampCacheSize: TEST_AUTH_CACHE_SIZE
+      })
       const sk = testPairs[1]
       const fileContents = sk.toWIF()
       const blob = Buffer.from(fileContents)
@@ -207,7 +241,7 @@ export function testHttpWithInMemoryDriver() {
       const authPart = auth.V1Authentication.makeAuthPart(sk, challenge)
       const authorization = `bearer ${authPart}`
 
-      const storeResponse = await request(app).post(path)
+      await request(app).post(path)
         .set('Content-Type', 'application/octet-stream')
         .set('Authorization', authorization)
         .send(blob)