Merge pull request #167 from adhocteam/cm-204-add-file-queue

Cm 204 add file queue components to backend app

.env.example

@@ -46,4 +46,5 @@ MINIO_ROOT_USER=EXAMPLEID
 # if using docker
 CLAMAV_ENDPOINT=http://clamav-rest:8080
 # if running locally
-# CLAMAV_ENDPOINT=http://localhost:8081
+# CLAMAV_ENDPOINT=http://localhost:8081
+REDIS_PASS=SUPERSECUREPASSWORD

docs/adr/0013-add-job-queue-and-worker.md

@@ -0,0 +1,26 @@
+# 13. Add job queue and worker
+
+# Date
+20201-02-13
+
+## Status
+
+Accepted
+
+## Context
+
+In order to satisfy the [RA-5](https://nvd.nist.gov/800-53/Rev4/control/RA-5)
+control around vulnerability scanning, we wish to scan all user-uploaded files
+with a malware detection service. We want to satisfy the following requirements.
+1. Scanning can be done asyncronously so as not to negatively impact the user experience.
+2. Scanning should be loosely coupled to main application to allow for more resiliance and fault tolerance.
+3. Scanning should be retried if malware detection service is unavailable.
+4. Scanning should run on a seperate instance to prevent a negative impact to the user experience.
+
+## Decision
+
+We will use redis as a queue and build a worker node which will take jobs from the queue, send them to the malware detection service and then update the database with the scan results.
+
+## Consequences
+
+All of the above requirements are filled. This does introduce some additional complexity in the form of a redis instance and worker instance. However, cloud.gov supplies managed redis instances and the worker will be built on top of an official node.js buildpack, so any additional maintenance from running these instances is negligable.

package.json

@@ -142,6 +142,7 @@
     "adm-zip": "^0.5.1",
     "aws-sdk": "^2.826.0",
     "axios": "^0.21.1",
+    "bull": "^3.20.1",
     "chromedriver": "^87.0.0",
     "client-oauth2": "^4.3.3",
     "cookie-session": "^1.4.0",

src/migrations/20210215161922-add-statuses-to-file-model.js

@@ -0,0 +1,12 @@
+module.exports = {
+  up: async (queryInterface) => {
+    await queryInterface.sequelize.query('ALTER TYPE "enum_Files_status" ADD VALUE \'QUEUEING_FAILED\';');
+    await queryInterface.sequelize.query('ALTER TYPE "enum_Files_status" ADD VALUE \'SCANNING_QUEUED\';');
+  },
+  down: async (queryInterface) => {
+    let query = 'DELETE FROM pg_enum WHERE enumlabel = \'QUEUEING_FAILED\' AND enumtypid = ( SELECT oid FROM pg_type WHERE typname = \'enum_Files_status\')';
+    await queryInterface.sequelize.query(query);
+    query = 'DELETE FROM pg_enum WHERE enumlabel = \'SCANNING_QUEUED\' AND enumtypid = ( SELECT oid FROM pg_type WHERE typname = \'enum_Files_status\')';
+    await queryInterface.sequelize.query(query);
+  },
+};

src/models/file.js

@@ -26,7 +26,16 @@ module.exports = (sequelize, DataTypes) => {
       allowNull: false,
     },
     status: {
-      type: DataTypes.ENUM('UPLOADING', 'UPLOADED', 'UPLOAD_FAILED', 'SCANNING', 'APPROVED', 'REJECTED'),
+      type: DataTypes.ENUM(
+        'UPLOADING',
+        'UPLOADED',
+        'UPLOAD_FAILED',
+        'QUEUEING_FAILED',
+        'SCANNING_QUEUED',
+        'SCANNING',
+        'APPROVED',
+        'REJECTED',
+      ),
       allowNull: false,
     },
     attachmentType: {

src/routes/files/handlers.js

@@ -3,10 +3,12 @@ import * as fs from 'fs';
 import handleErrors from '../../lib/apiErrorHandler';
 import { File } from '../../models';
 import s3Uploader from '../../lib/s3Uploader';
+import addToScanQueue from '../../services/queue';
 
 import ActivityReportPolicy from '../../policies/activityReport';
 import { activityReportById } from '../../services/activityReports';
 import { userById } from '../../services/users';
+import { auditLogger } from '../../logger';
 
 const fileType = require('file-type');
 const multiparty = require('multiparty');
@@ -20,14 +22,25 @@ const logContext = {
   namespace,
 };
 
+const fileStatuses = {
+  uploading: 'UPLOADING',
+  uploaded: 'UPLOADED',
+  uploadFailed: 'UPLOAD_FAILED',
+  queued: 'SCANNING_QUEUED',
+  queueingFailed: 'QUEUEING_FAILED',
+  scanning: 'SCANNING',
+  approved: 'APPROVED',
+  rejected: 'REJECTED',
+};
+
 export const createFileMetaData = async (
   originalFileName, s3FileName, reportId, attachmentType, fileSize) => {
   const newFile = {
     activityReportId: reportId,
     originalFileName,
     attachmentType,
     key: s3FileName,
-    status: 'UPLOADING',
+    status: fileStatuses.uploading,
     fileSize,
   };
   let file;
@@ -111,13 +124,23 @@ export default async function uploadHandler(req, res) {
     }
     try {
       await s3Uploader(buffer, fileName, type);
-      await updateStatus(metadata.id, 'UPLOADED');
+      await updateStatus(metadata.id, fileStatuses.uploaded);
       res.status(200).send({ id: metadata.id });
     } catch (err) {
       if (metadata) {
-        await updateStatus(metadata.id, 'UPLOAD_FAILED');
+        await updateStatus(metadata.id, fileStatuses.uploadFailed);
       }
       await handleErrors(req, res, err, logContext);
+      return;
+    }
+    try {
+      await addToScanQueue({ key: metadata.key });
+      await updateStatus(metadata.id, fileStatuses.queued);
+    } catch (err) {
+      if (metadata) {
+        await updateStatus(metadata.id, fileStatuses.queueingFailed);
+        auditLogger.error(`${logContext} Failed to queue ${metadata.originalFileName}. Error: ${err}`);
+      }
     }
   });
 }

src/routes/files/handlers.test.js

@@ -7,6 +7,7 @@ import db, {
 } from '../../models';
 import app from '../../app';
 import s3Uploader from '../../lib/s3Uploader';
+import * as queue from '../../services/queue';
 import SCOPES from '../../middleware/scopeConstants';
 import { REPORT_STATUSES } from '../../constants';
 import ActivityReportPolicy from '../../policies/activityReport';
@@ -43,6 +44,7 @@ const mockUser = {
 
 const mockSession = jest.fn();
 mockSession.userId = mockUser.id;
+const mockAddToScanQueue = jest.spyOn(queue, 'default').mockImplementation(() => jest.fn());
 
 const reportObject = {
   activityRecipientType: 'grantee',
@@ -93,6 +95,7 @@ describe('File Upload', () => {
           fileId = res.body.id;
           expect(s3Uploader).toHaveBeenCalled();
         });
+      expect(mockAddToScanQueue).toHaveBeenCalled();
     });
     it('checks the metadata was uploaded to the database', async () => {
       ActivityReportPolicy.mockImplementation(() => ({

src/services/queue.js

@@ -0,0 +1,10 @@
+import Queue from 'bull';
+
+const REDIS_PORT = process.env.REDIS_PORT || 6379;
+const { REDIS_HOST, REDIS_PASS } = process.env;
+
+const scanQueue = new Queue('scan', `redis://${REDIS_HOST}:${REDIS_PORT}`, { redis: { password: REDIS_PASS } });
+
+export default async function addToScanQueue(fileKey) {
+  await scanQueue.add(fileKey);
+}