Skip to content

Commit d26e529

Browse files
ngraefkylefarris
ngraef
authored and
kylefarris
committed
fix passthrough for node 16 by not returning a Promise from transform
1 parent c1d4f4c commit d26e529

2 files changed

Lines changed: 150 additions & 120 deletions

File tree

‎index.js‎

Lines changed: 133 additions & 120 deletions
Original file line numberDiff line numberDiff line change
@@ -1182,7 +1182,7 @@ class NodeClam {
11821182
// Ex. uploadStream.pipe(<this_transform_stream>).pipe(destination_stream)
11831183
return new Transform({
11841184
// This should be fired on each chunk received
1185-
async transform(chunk, encoding, cb) {
1185+
transform(chunk, encoding, cb) {
11861186
// DRY method for handling each chunk as it comes in
11871187
const doTransform = () => {
11881188
// Write data to our fork stream. If it fails,
@@ -1231,134 +1231,147 @@ class NodeClam {
12311231
// Setup an array to collect the responses from ClamAV
12321232
this._clamavResponseChunks = [];
12331233

1234-
try {
1235-
// Get a connection to the ClamAV Socket
1236-
this._clamavSocket = await me._initSocket('passthrough');
1237-
if (me.settings.debugMode) console.log(`${me.debugLabel}: ClamAV Socket Initialized...`);
1238-
1239-
// Setup a pipeline that will pass chunks through our custom Tranform and on to ClamAV
1240-
this._forkStream.pipe(this._clamavTransform).pipe(this._clamavSocket);
1241-
1242-
// When the CLamAV socket connection is closed (could be after 'end' or because of an error)...
1243-
this._clamavSocket
1244-
.on('close', (hadError) => {
1245-
if (me.settings.debugMode)
1246-
console.log(
1247-
`${me.debugLabel}: ClamAV socket has been closed! Because of Error:`,
1248-
hadError
1249-
);
1250-
this._clamavSocket.end();
1251-
})
1252-
// When the ClamAV socket connection ends (receives chunk)
1253-
.on('end', () => {
1254-
this._clamavSocket.end();
1255-
if (me.settings.debugMode)
1256-
console.log(`${me.debugLabel}: ClamAV socket has received the last chunk!`);
1257-
// Process the collected chunks
1258-
const response = Buffer.concat(this._clamavResponseChunks);
1259-
const result = me._processResult(response.toString('utf8'), null);
1260-
this._clamavResponseChunks = [];
1261-
if (me.settings.debugMode) {
1262-
console.log(`${me.debugLabel}: Result of scan:`, result);
1263-
console.log(
1264-
`${me.debugLabel}: It took ${_avScanTime} seconds to scan the file(s).`
1265-
);
1266-
clearScanBenchmark();
1267-
}
1234+
// Get a connection to the ClamAV Socket
1235+
me._initSocket('passthrough').then(
1236+
(socket) => {
1237+
this._clamavSocket = socket;
1238+
1239+
if (me.settings.debugMode) console.log(`${me.debugLabel}: ClamAV Socket Initialized...`);
12681240

1269-
// If the scan timed-out
1270-
if (result.timeout === true) this.emit('timeout');
1241+
// Setup a pipeline that will pass chunks through our custom Tranform and on to ClamAV
1242+
this._forkStream.pipe(this._clamavTransform).pipe(this._clamavSocket);
12711243

1272-
// NOTE: "scan-complete" could be called by the `handleError` method.
1273-
// We don't want to to double-emit this message.
1274-
if (_scanComplete === false) {
1275-
_scanComplete = true;
1244+
// When the CLamAV socket connection is closed (could be after 'end' or because of an error)...
1245+
this._clamavSocket
1246+
.on('close', (hadError) => {
1247+
if (me.settings.debugMode)
1248+
console.log(
1249+
`${me.debugLabel}: ClamAV socket has been closed! Because of Error:`,
1250+
hadError
1251+
);
12761252
this._clamavSocket.end();
1277-
this.emit('scan-complete', result);
1278-
}
1279-
})
1280-
// If connection timesout.
1281-
.on('timeout', () => {
1282-
this.emit('timeout', new Error('Connection to host/socket has timed out'));
1283-
this._clamavSocket.end();
1284-
if (me.settings.debugMode)
1285-
console.log(`${me.debugLabel}: Connection to host/socket has timed out`);
1286-
})
1287-
// When the ClamAV socket is ready to receive packets (this will probably never fire here)
1288-
.on('ready', () => {
1289-
if (me.settings.debugMode)
1290-
console.log(`${me.debugLabel}: ClamAV socket ready to receive`);
1291-
})
1292-
// When we are officially connected to the ClamAV socket (probably will never fire here)
1293-
.on('connect', () => {
1294-
if (me.settings.debugMode) console.log(`${me.debugLabel}: Connected to ClamAV socket`);
1295-
})
1296-
// If an error is emitted from the ClamAV socket
1297-
.on('error', (err) => {
1298-
console.error(`${me.debugLabel}: Error emitted from ClamAV socket: `, err);
1299-
handleError(err);
1300-
})
1301-
// If ClamAV is sending stuff to us (ie, an "OK", "Virus FOUND", or "ERROR")
1302-
.on('data', (cvChunk) => {
1303-
// Push this chunk to our results collection array
1304-
this._clamavResponseChunks.push(cvChunk);
1305-
if (me.settings.debugMode)
1306-
console.log(`${me.debugLabel}: Got result!`, cvChunk.toString());
1307-
1308-
// Parse what we've gotten back from ClamAV so far...
1309-
const response = Buffer.concat(this._clamavResponseChunks);
1310-
const result = me._processResult(response.toString(), null);
1311-
1312-
// If there's an error supplied or if we detect a virus or timeout, stop stream immediately.
1313-
if (
1314-
result instanceof NodeClamError ||
1315-
(typeof result === 'object' &&
1316-
(('isInfected' in result && result.isInfected === true) ||
1317-
('timeout' in result && result.timeout === true)))
1318-
) {
1319-
// If a virus is detected...
1320-
if (
1321-
typeof result === 'object' &&
1322-
'isInfected' in result &&
1323-
result.isInfected === true
1324-
) {
1325-
handleError(null, true, result);
1253+
})
1254+
// When the ClamAV socket connection ends (receives chunk)
1255+
.on('end', () => {
1256+
this._clamavSocket.end();
1257+
if (me.settings.debugMode)
1258+
console.log(`${me.debugLabel}: ClamAV socket has received the last chunk!`);
1259+
// Process the collected chunks
1260+
const response = Buffer.concat(this._clamavResponseChunks);
1261+
const result = me._processResult(response.toString('utf8'), null);
1262+
this._clamavResponseChunks = [];
1263+
if (me.settings.debugMode) {
1264+
console.log(`${me.debugLabel}: Result of scan:`, result);
1265+
console.log(
1266+
`${me.debugLabel}: It took ${_avScanTime} seconds to scan the file(s).`
1267+
);
1268+
clearScanBenchmark();
13261269
}
13271270

1328-
// If a timeout is detected...
1329-
else if (
1330-
typeof result === 'object' &&
1331-
'isInfected' in result &&
1332-
result.isInfected === true
1333-
) {
1334-
this.emit('timeout');
1335-
handleError(null, false, result);
1336-
}
1271+
// If the scan timed-out
1272+
if (result.timeout === true) this.emit('timeout');
13371273

1338-
// If any other kind of error is detected...
1339-
else {
1340-
handleError(result);
1274+
// NOTE: "scan-complete" could be called by the `handleError` method.
1275+
// We don't want to to double-emit this message.
1276+
if (_scanComplete === false) {
1277+
_scanComplete = true;
1278+
this._clamavSocket.end();
1279+
this.emit('scan-complete', result);
13411280
}
1342-
}
1343-
// For debugging purposes, spit out what was processed (if anything).
1344-
else if (me.settings.debugMode)
1345-
console.log(`${me.debugLabel}: Processed Result: `, result, response.toString());
1346-
});
1281+
})
1282+
// If connection timesout.
1283+
.on('timeout', () => {
1284+
this.emit('timeout', new Error('Connection to host/socket has timed out'));
1285+
this._clamavSocket.end();
1286+
if (me.settings.debugMode)
1287+
console.log(`${me.debugLabel}: Connection to host/socket has timed out`);
1288+
})
1289+
// When the ClamAV socket is ready to receive packets (this will probably never fire here)
1290+
.on('ready', () => {
1291+
if (me.settings.debugMode)
1292+
console.log(`${me.debugLabel}: ClamAV socket ready to receive`);
1293+
})
1294+
// When we are officially connected to the ClamAV socket (probably will never fire here)
1295+
.on('connect', () => {
1296+
if (me.settings.debugMode)
1297+
console.log(`${me.debugLabel}: Connected to ClamAV socket`);
1298+
})
1299+
// If an error is emitted from the ClamAV socket
1300+
.on('error', (err) => {
1301+
console.error(`${me.debugLabel}: Error emitted from ClamAV socket: `, err);
1302+
handleError(err);
1303+
})
1304+
// If ClamAV is sending stuff to us (ie, an "OK", "Virus FOUND", or "ERROR")
1305+
.on('data', (cvChunk) => {
1306+
// Push this chunk to our results collection array
1307+
this._clamavResponseChunks.push(cvChunk);
1308+
if (me.settings.debugMode)
1309+
console.log(`${me.debugLabel}: Got result!`, cvChunk.toString());
1310+
1311+
// Parse what we've gotten back from ClamAV so far...
1312+
const response = Buffer.concat(this._clamavResponseChunks);
1313+
const result = me._processResult(response.toString(), null);
1314+
1315+
// If there's an error supplied or if we detect a virus or timeout, stop stream immediately.
1316+
if (
1317+
result instanceof NodeClamError ||
1318+
(typeof result === 'object' &&
1319+
(('isInfected' in result && result.isInfected === true) ||
1320+
('timeout' in result && result.timeout === true)))
1321+
) {
1322+
// If a virus is detected...
1323+
if (
1324+
typeof result === 'object' &&
1325+
'isInfected' in result &&
1326+
result.isInfected === true
1327+
) {
1328+
handleError(null, true, result);
1329+
}
1330+
1331+
// If a timeout is detected...
1332+
else if (
1333+
typeof result === 'object' &&
1334+
'isInfected' in result &&
1335+
result.isInfected === true
1336+
) {
1337+
this.emit('timeout');
1338+
handleError(null, false, result);
1339+
}
1340+
1341+
// If any other kind of error is detected...
1342+
else {
1343+
handleError(result);
1344+
}
1345+
}
1346+
// For debugging purposes, spit out what was processed (if anything).
1347+
else if (me.settings.debugMode)
1348+
console.log(
1349+
`${me.debugLabel}: Processed Result: `,
1350+
result,
1351+
response.toString()
1352+
);
1353+
});
1354+
1355+
if (me.settings.debugMode) console.log(`${me.debugLabel}: Doing initial transform!`);
1356+
// Handle the chunk
1357+
doTransform();
1358+
},
1359+
(err) => {
1360+
// Close socket if it's currently valid
1361+
if (
1362+
this._clamavSocket &&
1363+
'readyState' in this._clamavSocket &&
1364+
this._clamavSocket.readyState
1365+
) {
1366+
this._clamavSocket.end();
1367+
}
13471368

1348-
if (me.settings.debugMode) console.log(`${me.debugLabel}: Doing initial transform!`);
1349-
// Handle the chunk
1350-
doTransform();
1351-
} catch (err) {
1352-
// Close socket if it's currently valid
1353-
if (this._clamavSocket && 'readyState' in this._clamavSocket && this._clamavSocket.readyState) {
1354-
this._clamavSocket.end();
1369+
// If there's an issue connecting to the ClamAV socket, this is where that's handled
1370+
if (me.settings.debugMode)
1371+
console.error(`${me.debugLabel}: Error initiating socket to ClamAV: `, err);
1372+
handleError(err);
13551373
}
1356-
1357-
// If there's an issue connecting to the ClamAV socket, this is where that's handled
1358-
if (me.settings.debugMode)
1359-
console.error(`${me.debugLabel}: Error initiating socket to ClamAV: `, err);
1360-
handleError(err);
1361-
}
1374+
);
13621375
} else {
13631376
// if (me.settings.debugMode) console.log(`${me.debugLabel}: Doing transform: ${++counter}`);
13641377
// Handle the chunk

‎tests/index.js‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1530,6 +1530,23 @@ describe('passthrough', () => {
15301530
});
15311531
});
15321532

1533+
// https://github.com/kylefarris/clamscan/issues/82
1534+
it('should not throw multiple callback error', (done) => {
1535+
// To reliably reproduce the issue in the broken code, it's important that this is an async generator
1536+
// and it emits some chunks larger than the default highWaterMark of 16 KB.
1537+
async function* gen(i = 10) {
1538+
while (i < 25) {
1539+
yield Buffer.from(new Array(i++ * 1024).fill());
1540+
}
1541+
}
1542+
1543+
const input = Readable.from(gen());
1544+
const av = clamscan.passthrough();
1545+
1546+
// The failure case will throw an error and not finish
1547+
input.pipe(av).on('end', done).resume();
1548+
});
1549+
15331550
if (!process.env.CI) {
15341551
it('should handle a 0-byte file', () => {
15351552
const input = fs.createReadStream(emptyFile);

0 commit comments

Comments
 (0)
⚡