Implement maxUploadFileSize

hub/CHANGELOG.md

@@ -17,6 +17,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   a `stat: true` option. 
 - Implemented `readFile` and `fileStat` methods on all drivers, however, 
   these are not yet in use or publicly exposed via any endpoints. 
+- The max file upload size is configurable and reported in `GET /hub_info`. 
 ### Changed
 - Concurrent requests to modify a resource using the `/store/${address}/...`
   or `/delete/${address}/...` endpoints are now always rejected with a 

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 { DriverModel, PerformWriteArgs, PerformRenameArgs, PerformDeleteArgs, PerformListFilesArgs, ListFilesStatResult, ListFilesResult } from './driverModel'
 import { HubConfigInterface } from './config'
-import { logger, generateUniqueID } from './utils'
+import { logger, generateUniqueID, bytesToMegabytes, megabytesToBytes, monitorStreamProgress } 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,29 @@ export class HubServer {
       }
     }
 
-    const readURL = await this.driver.performWrite(writeCommand)
+    // Create a PassThrough stream to monitor streaming chunk sizes. 
+    const { monitoredStream, pipelinePromise } = monitorStreamProgress(stream, totalBytes => {
+      if (totalBytes > contentLengthBytes) {
+        const errMsg = `Max file upload size is ${this.maxFileUploadSizeMB} megabytes. ` + 
+          `Rejected POST body stream of ${bytesToMegabytes(totalBytes, 4)} megabytes`
+        // Log error -- this situation is indicative of a malformed client request
+        // where the reported Content-Size is less than the upload size. 
+        logger.warn(`${errMsg}, address: ${address}`)
+
+        // Destroy the request stream -- cancels reading from the client
+        // and cancels uploading to the storage driver.
+        const error = new PayloadTooLargeError(errMsg)
+        stream.destroy(error)
+        throw error
+      }
+    })
+
+    const writeCommand: PerformWriteArgs = {
+      storageTopLevel: address,
+      path, stream: monitoredStream, contentType,
+      contentLength: contentLengthBytes
+    }
+    const [readURL] = await Promise.all([this.driver.performWrite(writeCommand), pipelinePromise])
     const driverPrefix = this.driver.getReadURLPrefix()
     const readURLPrefix = this.getReadURLPrefix()
     if (readURLPrefix !== driverPrefix && readURL.startsWith(driverPrefix)) {

hub/src/server/utils.ts

@@ -1,6 +1,6 @@
 
 
-import stream from 'stream'
+import stream, { Readable, PassThrough } from 'stream'
 import { DriverConstructor, DriverStatics } from './driverModel'
 import S3Driver from './drivers/S3Driver'
 import AzDriver from './drivers/AzDriver'
@@ -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)
 }
@@ -79,6 +87,59 @@ export function timeout(milliseconds: number): Promise<void> {
   })
 }
 
+export interface StreamProgressCallback {
+  /**
+   * A callback that is invoked each time a chunk passes through the stream. 
+   * This callback can throw an Error and it will be propagated 
+   * If this callback throws an error, it will be propagated through the stream
+   * pipeline. 
+   * @param totalBytes Total bytes read (includes the current chunk bytes). 
+   * @param chunkBytes Bytes read in the current chunk. 
+   */
+  (totalBytes: number, chunkBytes: number): void;
+}
+
+export interface MonitorStreamResult {
+  monitoredStream: Readable;
+  pipelinePromise: Promise<void>;
+}
+
+export function monitorStreamProgress(
+  inputStream: Readable, 
+  progressCallback: StreamProgressCallback
+): MonitorStreamResult {
+
+  // Create a PassThrough stream to monitor streaming chunk sizes. 
+  let monitoredContentSize = 0
+  const monitorStream = new PassThrough({
+    transform: (chunk: Buffer, _encoding, callback) => {
+      monitoredContentSize += chunk.length
+      try {
+        progressCallback(monitoredContentSize, chunk.length)
+        // Pass the chunk Buffer through, untouched. This takes the fast 
+        // path through the stream pipe lib. 
+        callback(null, chunk)
+      } catch (error) {
+        callback(error)
+      }
+    }
+  })
+
+  // Use the stream pipe API to monitor a stream with correct back pressure
+  // 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
+  //     https://nodejs.org/ja/docs/guides/backpressuring-in-streams/
+  const monitorPipeline = pipelineAsync(inputStream, monitorStream)
+
+  const result: MonitorStreamResult = {
+    monitoredStream: monitorStream,
+    pipelinePromise: monitorPipeline
+  }
+
+  return result
+}
+
 
 export class AsyncMutexScope {