Generic Attachment-Processing registration

Dockerfile

@@ -48,6 +48,37 @@ RUN cd ${MAGE_SERVER}/plugins/sftp/web \
     && cd ${MAGE_PACKAGES} \
     && npm pack ${MAGE_SERVER}/plugins/sftp/web/dist/main
 
+FROM $BASE_IMAGE AS build-clamav-service
+WORKDIR /clamav.service
+COPY plugins/clamav/service/ ./
+COPY --from=build-service /packages/ /packages
+RUN npm i /packages/ngageoint-mage.service-*.tgz
+RUN npm run build
+RUN npm pack
+
+FROM $BASE_IMAGE AS build-sftp-web
+WORKDIR /sftp.web
+COPY plugins/sftp/web/ ./
+COPY --from=build-web-app /packages/ /packages
+# TODO: uncomment this line to build with local core-lib when web plugin is upgraded to angular 20
+# RUN npm i /packages/ngageoint-mage.web-core-lib-*.tgz
+# TODO: remoove this line after activating above line
+RUN npm ci
+RUN npm run build
+RUN npm pack ./dist/main
+
+# Build instance
+FROM $BASE_IMAGE AS build-instance
+ENV MAGE_HOME=/home/mage/instance
+WORKDIR ${MAGE_HOME}
+COPY --from=build-service /packages/ngageoint-mage.service-*.tgz ${MAGE_HOME}/packages/
+COPY --from=build-web-app /packages/ngageoint-mage.web-app-*.tgz ${MAGE_HOME}/packages/
+COPY --from=build-image-service /image.service/ngageoint-mage.image.service-*.tgz ${MAGE_HOME}/packages/
+COPY --from=build-arcgis-service /arcgis.service/ngageoint-mage.*.tgz ${MAGE_HOME}/packages/
+COPY --from=build-arcgis-web /arcgis.web/ngageoint-mage.arcgis.web-*.tgz ${MAGE_HOME}/packages/
+COPY --from=build-sftp-service /sftp.service/ngageoint-*.tgz ${MAGE_HOME}/packages/
+COPY --from=build-sftp-web /sftp.web/ngageoint-*.tgz ${MAGE_HOME}/packages/
+COPY --from=build-clamav-service /clamav.service/ngageoint-*.tgz ${MAGE_HOME}/packages/
 RUN cd ${MAGE_SERVER}/plugins/arcgis/service \
     && npm link ../../../service \
     && npm run build \

docker-compose.yml

@@ -1,5 +1,16 @@
 services:
 
+  # Add ClamAV to build local container for testing
+  mage-clamav:
+    image: clamav/clamav:stable
+    platform: linux/amd64
+    volumes:
+      - ./clamav-test/data:/var/lib/clamav
+    ports:
+      - "3310:3310"
+    networks:
+      - mage.net
+
   mage-db:
     image: mongodb/mongodb-community-server:8.0-ubi9-slim
     # Uncomment the following ports block to allow the mongo client on your
@@ -34,6 +45,11 @@ services:
     environment:
       MAGE_MONGO_URL: mongodb://mage-db:27017/magedb
       MAGE_TOKEN_EXPIRATION: "28800"
+      # MAGE_PLUGINS: '{"servicePlugins":["@ngageoint/mage.sftp.service","@ngageoint/mage.arcgis.service","@ngageoint/mage.clamav.service"],"webUIPlugins":["@ngageoint/mage.sftp.web","@ngageoint/mage.arcgis.web-app"]}'
+      # Only used if @ngageoint/mage.clamav.service is enabled above - host/port
+      # for the mage-clamav container this compose file already defines.
+      MAGE_CLAMAV_HOST: mage-clamav
+      MAGE_CLAMAV_PORT: "3310"
       MAGE_PLUGINS: '{"servicePlugins":["@ngageoint/mage.sftp.service", "@ngageoint/mage.arcgis.service"],"webUIPlugins":["@ngageoint/mage.sftp.web", "@ngageoint/mage.arcgis.web-app"]}'
       MAGE_SFTP_KEY_FILE: /var/lib/mage/sftp_key
 

plugins/clamav/service/package.json

@@ -0,0 +1,36 @@
+{
+  "name": "@ngageoint/mage.clamav.service",
+  "version": "1.0.0-beta.1",
+  "description": "The ClamAV service package is a Mage server plugin that scans attachment content for viruses before it is made available.",
+  "main": "lib/index.js",
+  "scripts": {
+    "build": "tsc",
+    "test": "npm-run-all build test:run",
+    "test:run": "jasmine"
+  },
+  "repository": {
+    "type": "git",
+    "url": "git+https://github.com/ngageoint/mage-server.git"
+  },
+  "keywords": [
+    "mage",
+    "plugin",
+    "clamav",
+    "attachment-processing"
+  ],
+  "author": "National Geospatial-Intelligence Agency",
+  "license": "MIT",
+  "bugs": {
+    "url": "https://github.com/ngageoint/mage-server/issues"
+  },
+  "homepage": "https://github.com/ngageoint/mage-server#readme",
+  "devDependencies": {
+    "@types/jasmine": "^3.10.6",
+    "jasmine": "^6.3.0",
+    "npm-run-all": "^4.1.5",
+    "typescript": "^5.9.3"
+  },
+  "peerDependencies": {
+    "@ngageoint/mage.service": "^6.5.8 || ^6.6.0-beta"
+  }
+}

plugins/clamav/service/src/clamHook.ts

@@ -0,0 +1,24 @@
+import fs from 'fs'
+import { scan } from './scan'
+import { AttachmentHook } from '@ngageoint/mage.service/lib/plugins.api/plugins.api.attachments'
+
+// ClamAV hook start
+export const clamavHook: AttachmentHook = async (attachment, stagedFilePath) => {
+    const host = process.env.MAGE_CLAMAV_HOST
+    const port = process.env.MAGE_CLAMAV_PORT ? Number(process.env.MAGE_CLAMAV_PORT) : undefined
+
+    // Outcome from stream
+    const outcome = await scan(fs.createReadStream(stagedFilePath), { host, port })
+
+    // Message return for a rejected statement
+    if (outcome.outcome === 'reject' ) {
+        return { outcome: 'reject', reason: `This file/attachment was flagged as potentially malicious (${outcome.reason}) and could not be uploaded` }
+    }
+
+    // Message return for errors (not including malicious attachments)
+    if (outcome.outcome === 'error') {
+        return { outcome: 'error', error: new Error(`Virus scan could not be completed (${outcome.error.message})`, { cause: outcome.error }) }
+    }
+
+    return outcome
+}

plugins/clamav/service/src/index.ts

@@ -0,0 +1,11 @@
+import { InitPluginHook } from '@ngageoint/mage.service/lib/plugins.api'
+import { clamavHook } from './clamHook'
+
+const clamavPluginHooks: InitPluginHook = {
+    init: async () => {
+        return { attachmentHooks: [clamavHook] }
+    }
+}
+
+// One object for export
+export = clamavPluginHooks

plugins/clamav/service/src/scan.ts

@@ -0,0 +1,110 @@
+import net from 'net'
+import { AttachmentHookOutcome } from '@ngageoint/mage.service/lib/plugins.api/plugins.api.attachments'
+
+const DEFAULT_TIMEOUT_MS = 15_000
+
+export type ScanOptions = {
+    host?: string
+    port?: number
+    timeoutMs?: number
+    createConnection?: (options: { host: string, port: number }) => net.Socket
+}
+
+// Scan function
+export function scan(inputStream: NodeJS.ReadableStream & { destroy: (error?: Error) => void }, options: ScanOptions = {}): Promise<AttachmentHookOutcome> {
+  const {
+    host = 'localhost',
+    port = 3310,
+    timeoutMs = DEFAULT_TIMEOUT_MS,
+    createConnection = net.createConnection,
+  } = options
+
+  return new Promise<AttachmentHookOutcome>((resolve) => {
+    const socket = createConnection({ host, port })
+
+    // Pause immediately, before wiring up any listeners, so the command is
+    // always sent before any input bytes reach clamd, regardless of which
+    // async operation (socket connect vs. stream read) happens to finish first.
+    inputStream.pause()
+
+    const responseChunks: Buffer[] = []
+    let settled = false
+
+    const timeout = setTimeout(() => finish(errorOutcome(new Error('scan timed out'))), timeoutMs)
+
+    function finish(outcome: AttachmentHookOutcome) {
+      if (settled) return
+      settled = true
+      clearTimeout(timeout) 
+      inputStream.destroy()
+      socket.destroy()
+      resolve(outcome)
+    }
+
+    socket.on('error', (err: Error) => finish(errorOutcome(new Error(`socket error: ${err.message}`))))
+
+    socket.on('connect', () => {
+      socket.write('zINSTREAM\0')
+      inputStream.resume()
+    })
+
+    inputStream.on('error', (err: Error) => finish(errorOutcome(new Error(`input stream error: ${err.message}`))))
+
+    inputStream.on('data', (chunk: Buffer) => {
+      // Each chunk is preceded by its own length as a 4-byte big-endian
+      // unsigned integer, per the INSTREAM protocol.
+      const lenPrefix = Buffer.alloc(4)
+      lenPrefix.writeUInt32BE(chunk.length, 0)
+      const stillHasRoom = socket.write(Buffer.concat([lenPrefix, chunk]))
+
+      // Honor backpressure: pause reading more input until the socket
+      // signals it has caught up.
+      if (!stillHasRoom) {
+        inputStream.pause()
+        socket.once('drain', () => inputStream.resume())
+      }
+    })
+
+    inputStream.on('end', () => {
+      // A zero-length chunk tells clamd no more data is coming.
+      socket.write(Buffer.alloc(4))
+    })
+
+    socket.on('data', (chunk: Buffer) => responseChunks.push(chunk))
+
+    socket.on('end', () => {
+      // clamd's reply to a z-prefixed command (like zINSTREAM) is
+      // NUL-terminated, not newline-terminated - strip NUL bytes before
+      // trimming whitespace.
+      const response = Buffer.concat(responseChunks).toString().replace(/\0/g, '').trim()
+      finish(parseResponse(response))
+    })
+  })
+}
+
+// clamd's INSTREAM reply is always "stream: <result>":
+//   - "stream: OK" - clean
+//   - "stream: <SignatureName> FOUND" - infected
+//   - anything else is treated as an error rather than guessed at
+function parseResponse(response: string): AttachmentHookOutcome {
+  const match = /^stream:\s*(.*)$/.exec(response)
+  if (!match) {
+    return errorOutcome(new Error(`unrecognized clamd response: "${response}"`))
+  }
+
+  const result = match[1].trim()
+  if (result === 'OK') {
+    return { outcome: 'pass' }
+  }
+
+  const foundMatch = /^(.+) FOUND$/.exec(result)
+  if (foundMatch) {
+    return { outcome: 'reject', reason: foundMatch[1] }
+  }
+
+  return errorOutcome(new Error(`unrecognized clamd result: "${result}"`))
+}
+
+function errorOutcome(error: Error): AttachmentHookOutcome {
+  return { outcome: 'error', error }
+}

plugins/clamav/service/tsconfig.json

@@ -0,0 +1,17 @@
+{
+  "include": ["src/**/*.ts", "src/**/*.spec.ts", "src/**/*.test.ts"],
+  "exclude": ["node_modules", "lib", "dist"],
+  "compilerOptions": {
+    "strict": true,
+    "outDir": "lib",
+    "target": "ES2022",
+    "lib": ["ES2022"],
+    "sourceMap": true,
+    "module": "commonjs",
+    "moduleResolution": "node",
+    "esModuleInterop": true,
+    "baseUrl": ".",
+    "types": ["jasmine", "node"],
+    "skipLibCheck": true
+  }
+}

plugins/image/service/src/processor.spec.ts

@@ -945,6 +945,15 @@ class BufferAttachmentStore implements AttachmentStore {
     this.pendingContent.set(id, Buffer.alloc(0))
     return pending
   }
+  async stagedContentPath(stagedContentId: StagedAttachmentContentId): Promise<string | AttachmentStoreError> {
+    if (!this.pendingContent.has(stagedContentId as string)) {
+      return new AttachmentStoreError(AttachmentStoreErrorCode.ContentNotFound, `pending content not found: ${stagedContentId}`)
+    }
+    return stagedContentId as string
+  }
+  async deleteStagedContent(stagedContentId: StagedAttachmentContentId): Promise<void | AttachmentStoreError> {
+    this.pendingContent.delete(stagedContentId as string)
+  }
   async saveContent(content: NodeJS.ReadableStream | StagedAttachmentContent, attachmentId: string, observation: Observation): Promise<AttachmentStoreError | AttachmentContentPatchAttrs | null> {
     if (typeof content !== 'string') {
       return new AttachmentStoreError(AttachmentStoreErrorCode.ContentNotFound, 'this store supports saving only staged content')

service/src/adapters/observations/adapters.observations.attachment_store.file_system.ts

@@ -3,7 +3,7 @@ import path from 'path'
 import stream from 'stream'
 import util from 'util'
 import uniqid from 'uniqid'
-import { Attachment, AttachmentStore, AttachmentStoreError, AttachmentStoreErrorCode, copyThumbnailAttrs, Observation, patchAttachment, StagedAttachmentContent, StagedAttachmentContentRef, Thumbnail, AttachmentContentPatchAttrs, ThumbnailContentPatchAttrs, AttachmentPatchAttrs } from '../../entities/observations/entities.observations'
+import { Attachment, AttachmentStore, AttachmentStoreError, AttachmentStoreErrorCode, copyThumbnailAttrs, Observation, patchAttachment, StagedAttachmentContent, StagedAttachmentContentRef, Thumbnail, AttachmentContentPatchAttrs, ThumbnailContentPatchAttrs, AttachmentPatchAttrs, StagedAttachmentContentId } from '../../entities/observations/entities.observations'
 import mime from 'mime-types'
 
 export class FileSystemAttachmentStore implements AttachmentStore {
@@ -21,6 +21,28 @@ export class FileSystemAttachmentStore implements AttachmentStore {
     return new StagedAttachmentContent(id, tempLocation)
   }
 
+  // Define the staged content path
+  async stagedContentPath(stagedContentId: StagedAttachmentContentId): Promise<string | AttachmentStoreError> {
+    return path.join(this.pendingDirPath, stagedContentId as string)
+  }
+
+  // Define delete staged content
+  async deleteStagedContent(stagedContentId: StagedAttachmentContentId): Promise<void | AttachmentStoreError> {
+    const tempPath = path.join(this.pendingDirPath, stagedContentId as string)
+    try{
+      // Cleanly removing path
+      await util.promisify(fs.rm)(tempPath)
+
+    } catch (err) {
+      // Message for crashes or any failures
+      const message = `error deleting staged content ${stagedContentId}`
+
+      // Add console output
+      console.error(message, err)
+      return new AttachmentStoreError(AttachmentStoreErrorCode.StorageError, `${message}: ${err instanceof Error ? err.message : String(err)}`)
+    }
+  }
+
   async saveContent(content: StagedAttachmentContentRef | NodeJS.ReadableStream, attachmentId: string, observation: Observation): Promise<null | AttachmentContentPatchAttrs | AttachmentStoreError> {
     const attachment = observation.attachmentFor(attachmentId)
     if (!attachment) {

service/src/adapters/observations/adapters.observations.db.mongoose.ts

@@ -146,6 +146,13 @@ export function docToEntity(doc: legacy.ObservationDocument, eventId: MageEventI
   return createDocumentMapping(eventId)(doc)
 }
 
+// Define what a pending attachment looks like
+export type PendingAttachmentReference = {
+  eventId: MageEventId
+  observationId: ObservationId
+  attachmentId: AttachmentId
+}
+
 function createDocumentMapping(eventId: MageEventId): DocumentMapping<legacy.ObservationDocument, ObservationAttrs> {
   return doc => {
     const attrs: ObservationAttrs = {
@@ -205,6 +212,11 @@ function attachmentAttrsForDoc(doc: legacy.AttachmentDocument): Attachment {
     oriented: doc.oriented,
     contentLocator: doc.relativePath,
     thumbnails: doc.thumbnails.map(thumbnailAttrsForDoc),
+    processingStatus: doc.processingStatus,
+    processingMessage: doc.processingMessage,
+    processingHook: doc.processingHook,
+    stagedContentId: doc.stagedContentId,
+    processingRetryCount: doc.processingRetryCount,
   }
 }
 
@@ -266,3 +278,42 @@ function thumbnailDocSeedForEntity(attrs: Thumbnail): legacy.ThumbnailDocAttrs {
     height: attrs.height,
   }
 }
+
+// Define the function to check for pending attachments
+export async function findPendingAttachments(limit: number): Promise<PendingAttachmentReference[]> {
+
+  // Set the array of documents
+  const eventDocs = await mongoose.connection.collection<{ _id: number, collectionName: string }>('events').find({}).toArray()
+
+  // Loop through each document
+  const references: PendingAttachmentReference[] = []
+  for (const eventDoc of eventDocs) {
+    if (references.length >= limit) {
+      break
+    }
+    const model = legacy.observationModel({ id: eventDoc._id, collectionName: eventDoc.collectionName })
+
+    // mongo aggregation pipeline
+    const remaining = limit - references.length
+    const pipeline = [
+      { $match: { 'attachments.processingStatus': 'pending' } },
+      { $unwind: '$attachments' },
+      { $match: { 'attachments.processingStatus': 'pending' } },
+      { $project: { _id: false, observationId: '$_id', attachmentId: '$attachments._id' } },
+      { $limit: remaining },
+    ]
+
+    // Run the pipeline against the model
+    const matches = await model.aggregate(pipeline)
+
+    // loop over the matches
+    for (const match of matches) {
+      references.push({
+        eventId: eventDoc._id,
+        observationId: match.observationId.toString(),
+        attachmentId: match.attachmentId.toString()
+      })
+    }
+  } 
+  return references
+}