Implement network scanning enhancements in SocketClient and WebSocketScanner
All checks were successful
farmcontrol/farmcontrol-server/pipeline/head This commit looks good

- Added support for multiple scan sessions in `SocketClient`, allowing concurrent network scans with unique session IDs.
- Enhanced `WebSocketScanner` to support continuous scanning with the option to stop scans gracefully.
- Introduced detailed event emissions for scan progress and completion, improving real-time feedback during network scans.
- Updated `handleScanNetworkStart` and `handleScanNetworkStop` methods to manage scan sessions effectively, including error handling and session cleanup.
- Improved logging for better visibility into scanning operations and results.
This commit is contained in:
Tom Butcher 2026-09-14 22:50:40 +01:00
parent e7bf650e82
commit 3505121f3e
4 changed files with 357 additions and 134 deletions

View File

@ -172,8 +172,16 @@ export class HostManager {
logger.error("Failed to reload printers:", error); logger.error("Failed to reload printers:", error);
}); });
return; return;
case "scanNetwork":
this.socketClient.handleScanNetworkStart(action.data, callback);
return;
case "scanNetworkStop":
this.socketClient.handleScanNetworkStop(action.data, callback);
return;
}
if (typeof callback === "function") {
callback({ success: true });
} }
callback({ success: true });
} }
async handleUpdate(data) { async handleUpdate(data) {

View File

@ -11,56 +11,84 @@ export class WebSocketScanner extends EventEmitter {
constructor(options = {}) { constructor(options = {}) {
super(); super();
this.scanning = false; this.scanning = false;
this.stopped = false;
this.workers = []; this.workers = [];
// Default to number of CPU cores, but allow override
this.maxThreads = options.maxThreads || os.cpus().length; this.maxThreads = options.maxThreads || os.cpus().length;
this.totalIPs = 0; this.totalIPs = 0;
this.scannedIPs = 0; this.scannedIPs = 0;
} }
/** /**
* Scans the local network for websocket services on a specified port * Scans the local network for services on a specified port.
* @param {number} port - Port number to scan * When loop is true, repeats until stopScan() is called.
* @returns {Promise<Array>} Array of IP addresses where websocket service was found
*/ */
async scanNetwork(port, protocol) { async scanNetwork(port, protocol, objectType = 'printer', { loop = false } = {}) {
// Clean up any existing workers before starting a new scan this.stopped = false;
this.cleanupWorkers(); this.scanning = true;
console.log("Cleaned up workers"); let foundServices = [];
if (this.scanning) { try {
throw new Error('Scan already in progress'); do {
await this.cleanupWorkers();
if (this.stopped) {
break;
}
foundServices = await this.scanOnce(port, protocol, objectType);
if (this.stopped) {
break;
}
this.emit('scanPassComplete', foundServices);
} while (loop && !this.stopped);
} finally {
await this.cleanupWorkers();
this.scanning = false;
} }
this.scanning = true; return foundServices;
}
async scanOnce(port, protocol, objectType) {
this.scannedIPs = 0; this.scannedIPs = 0;
const foundServices = []; const foundServices = [];
// Get local network range
const { startIP, endIP } = this.getLocalNetworkRange(); const { startIP, endIP } = this.getLocalNetworkRange();
const start = this.ipToNumber(startIP); const start = this.ipToNumber(startIP);
const end = this.ipToNumber(endIP); const end = this.ipToNumber(endIP);
// Calculate IP ranges for each worker
this.totalIPs = end - start + 1; this.totalIPs = end - start + 1;
const ipsPerWorker = Math.ceil(this.totalIPs / this.maxThreads); const ipsPerWorker = Math.ceil(this.totalIPs / this.maxThreads);
const workerPromises = []; const workerPromises = [];
for (let i = 0; i < this.maxThreads; i++) { for (let i = 0; i < this.maxThreads; i++) {
if (this.stopped) {
break;
}
const workerStart = start + (i * ipsPerWorker); const workerStart = start + (i * ipsPerWorker);
const workerEnd = Math.min(workerStart + ipsPerWorker - 1, end); const workerEnd = Math.min(workerStart + ipsPerWorker - 1, end);
if (workerStart > end) break;
const workerStartIP = this.numberToIP(workerStart); if (workerStart > end) break;
const workerEndIP = this.numberToIP(workerEnd);
const worker = new Worker(path.join(__dirname, 'websocketScannerWorker.js')); const worker = new Worker(path.join(__dirname, 'websocketScannerWorker.js'));
this.workers.push(worker); this.workers.push(worker);
console.log("Created worker", i);
const workerPromise = new Promise((resolve) => { const workerPromise = new Promise((resolve) => {
let settled = false;
const finish = (results) => {
if (settled) return;
settled = true;
resolve(results || []);
};
worker.on('message', (message) => { worker.on('message', (message) => {
if (this.stopped) {
finish([]);
return;
}
switch (message.type) { switch (message.type) {
case 'serviceFound': case 'serviceFound':
foundServices.push({ ip: message.ip, hostname: message.hostname }); foundServices.push({ ip: message.ip, hostname: message.hostname });
@ -75,44 +103,41 @@ export class WebSocketScanner extends EventEmitter {
}); });
break; break;
case 'scanComplete': case 'scanComplete':
resolve(message.results); finish(message.results);
break; break;
} }
}); });
worker.on('error', () => finish([]));
worker.on('exit', () => finish([]));
}); });
worker.postMessage({ worker.postMessage({
type: 'scan', type: 'scan',
startIP: workerStartIP, startIP: this.numberToIP(workerStart),
endIP: workerEndIP, endIP: this.numberToIP(workerEnd),
port, port,
protocol protocol,
objectType
}); });
workerPromises.push(workerPromise); workerPromises.push(workerPromise);
} }
await Promise.all(workerPromises); await Promise.all(workerPromises);
this.cleanupWorkers(); await this.cleanupWorkers();
this.scanning = false;
return foundServices; return foundServices;
} }
cleanupWorkers() { async cleanupWorkers() {
this.workers.forEach(worker => worker.terminate()); const workers = this.workers.splice(0, this.workers.length);
this.workers = []; await Promise.all(workers.map((worker) => worker.terminate()));
} }
/**
* Gets the local network IP range
* @returns {Object} Object containing startIP and endIP
*/
getLocalNetworkRange() { getLocalNetworkRange() {
const interfaces = os.networkInterfaces(); const interfaces = os.networkInterfaces();
let localIP = null; let localIP = null;
let subnetMask = null; let subnetMask = null;
// Find the first non-internal IPv4 address
for (const name of Object.keys(interfaces)) { for (const name of Object.keys(interfaces)) {
for (const iface of interfaces[name]) { for (const iface of interfaces[name]) {
if (iface.family === 'IPv4' && !iface.internal) { if (iface.family === 'IPv4' && !iface.internal) {
@ -128,39 +153,22 @@ export class WebSocketScanner extends EventEmitter {
throw new Error('Could not determine local network IP address'); throw new Error('Could not determine local network IP address');
} }
// Convert IP and subnet mask to numbers
const ipNum = this.ipToNumber(localIP); const ipNum = this.ipToNumber(localIP);
const maskNum = this.ipToNumber(subnetMask); const maskNum = this.ipToNumber(subnetMask);
// Calculate network address
const networkNum = ipNum & maskNum; const networkNum = ipNum & maskNum;
// Calculate broadcast address
const broadcastNum = networkNum | (~maskNum >>> 0); const broadcastNum = networkNum | (~maskNum >>> 0);
// Start IP is network address + 1 return {
const startIP = this.numberToIP(networkNum + 1); startIP: this.numberToIP(networkNum + 1),
// End IP is broadcast address - 1 endIP: this.numberToIP(broadcastNum - 1)
const endIP = this.numberToIP(broadcastNum - 1); };
return { startIP, endIP };
} }
/**
* Converts an IP address to a number
* @param {string} ip - IP address to convert
* @returns {number} Numeric representation of IP
*/
ipToNumber(ip) { ipToNumber(ip) {
return ip.split('.') return ip.split('.')
.reduce((acc, octet) => (acc << 8) + parseInt(octet), 0) >>> 0; .reduce((acc, octet) => (acc << 8) + parseInt(octet), 0) >>> 0;
} }
/**
* Converts a number to an IP address
* @param {number} num - Number to convert
* @returns {string} IP address
*/
numberToIP(num) { numberToIP(num) {
return [ return [
(num >>> 24) & 255, (num >>> 24) & 255,
@ -170,16 +178,9 @@ export class WebSocketScanner extends EventEmitter {
].join('.'); ].join('.');
} }
/** async stopScan() {
* Stops the current scan this.stopped = true;
*/
stopScan() {
this.scanning = false; this.scanning = false;
this.cleanupWorkers(); await this.cleanupWorkers();
} }
} }
// To stop scanning at any time:
// scanner.stopScan();

View File

@ -1,48 +1,51 @@
import { parentPort, workerData } from 'worker_threads'; import { parentPort } from 'worker_threads';
import WebSocket from 'ws'; import WebSocket from 'ws';
import net from 'net';
import http from 'http';
import https from 'https';
import { loadConfig } from "../config.js"; import { loadConfig } from "../config.js";
import log4js from "log4js"; import log4js from "log4js";
import dns from 'dns'; import dns from 'dns';
// Load configuration
const config = loadConfig(); const config = loadConfig();
class WebSocketScannerWorker { class WebSocketScannerWorker {
constructor() { constructor() {
this.randomId = Math.floor(Math.random() * 1000).toString().padStart(3, '0'); this.randomId = Math.floor(Math.random() * 1000).toString().padStart(3, '0');
this.logger = log4js.getLogger(`WS Scanner #${this.randomId}`); this.logger = log4js.getLogger(`WS Scanner #${this.randomId}`);
this.logger.level = config.server.logLevel; this.logger.level = config.logLevel;
this.timeout = 2000; // 2 second timeout for each connection attempt this.timeout = 2000;
} }
async scanRange(startIP, endIP, port, protocol) { async scanRange(startIP, endIP, port, protocol, objectType) {
const start = this.ipToNumber(startIP); const start = this.ipToNumber(startIP);
const end = this.ipToNumber(endIP); const end = this.ipToNumber(endIP);
const foundServices = []; const foundServices = [];
this.logger.info(`Scanning ${startIP} - ${endIP} on port: ${port} using: ${protocol}`); this.logger.info(
`Scanning ${startIP} - ${endIP} on port: ${port} using: ${protocol} objectType: ${objectType}`
);
for (let ip = start; ip <= end; ip++) { for (let ip = start; ip <= end; ip++) {
const currentIP = this.numberToIP(ip); const currentIP = this.numberToIP(ip);
const url = `${protocol}://${currentIP}:${port}/websocket`;
try { try {
this.logger.debug(`Checking ${currentIP} for websocket service on port ${port}`); this.logger.debug(`Checking ${currentIP} on port ${port}`);
const isOpen = await this.checkWebSocket(url); const isOpen = await this.probe(currentIP, port, protocol, objectType);
if (isOpen) { if (isOpen) {
const hostname = await this.resolveHostname(currentIP); const hostname = await this.resolveHostname(currentIP);
foundServices.push({ ip: currentIP, hostname }); foundServices.push({ ip: currentIP, hostname });
this.logger.info(`WebSocket connection successful for ${currentIP}${hostname ? ` (${hostname})` : ''}`); this.logger.info(
`Service found at ${currentIP}${hostname ? ` (${hostname})` : ''}`
);
parentPort.postMessage({ type: 'serviceFound', ip: currentIP, hostname }); parentPort.postMessage({ type: 'serviceFound', ip: currentIP, hostname });
} else {
this.logger.debug(`WebSocket connection failed for ${currentIP}`);
} }
} catch (error) { } catch (error) {
// Connection failed, continue scanning // Connection failed, continue scanning
} }
parentPort.postMessage({ parentPort.postMessage({
type: 'scanProgress', type: 'scanProgress',
currentIP, currentIP,
increment: 1 increment: 1
}); });
@ -51,12 +54,28 @@ class WebSocketScannerWorker {
return foundServices; return foundServices;
} }
async probe(ip, port, protocol, objectType) {
if (protocol === 'serial' || protocol === 'system') {
return false;
}
if (objectType === 'printer' || protocol === 'ws' || protocol === 'wss') {
const url = `${protocol || 'ws'}://${ip}:${port}/websocket`;
return this.checkWebSocket(url);
}
if (protocol === 'http' || protocol === 'https') {
return this.checkHttp(ip, port, protocol);
}
return this.checkTcp(ip, port);
}
async resolveHostname(ip) { async resolveHostname(ip) {
try { try {
const hostnames = await dns.promises.reverse(ip); const hostnames = await dns.promises.reverse(ip);
return hostnames[0] || null; return hostnames[0] || null;
} catch (error) { } catch (error) {
// Only log errors that aren't ENOTFOUND (which is expected for many IPs)
if (error.code !== 'ENOTFOUND') { if (error.code !== 'ENOTFOUND') {
this.logger.warn(`Unexpected error resolving hostname for ${ip}: ${error.message}`); this.logger.warn(`Unexpected error resolving hostname for ${ip}: ${error.message}`);
} }
@ -67,7 +86,7 @@ class WebSocketScannerWorker {
checkWebSocket(url) { checkWebSocket(url) {
return new Promise((resolve) => { return new Promise((resolve) => {
const ws = new WebSocket(url); const ws = new WebSocket(url);
let timeout = setTimeout(() => { const timeout = setTimeout(() => {
ws.terminate(); ws.terminate();
resolve(false); resolve(false);
}, this.timeout); }, this.timeout);
@ -85,6 +104,61 @@ class WebSocketScannerWorker {
}); });
} }
checkTcp(ip, port) {
return new Promise((resolve) => {
const socket = new net.Socket();
const timeout = setTimeout(() => {
socket.destroy();
resolve(false);
}, this.timeout);
socket.once('connect', () => {
clearTimeout(timeout);
socket.destroy();
resolve(true);
});
socket.once('error', () => {
clearTimeout(timeout);
socket.destroy();
resolve(false);
});
socket.connect(port, ip);
});
}
checkHttp(ip, port, protocol) {
const client = protocol === 'https' ? https : http;
return new Promise((resolve) => {
const req = client.request(
{
host: ip,
port,
method: 'GET',
path: '/',
timeout: this.timeout,
rejectUnauthorized: false
},
(res) => {
res.resume();
resolve(true);
}
);
req.on('timeout', () => {
req.destroy();
resolve(false);
});
req.on('error', () => {
resolve(false);
});
req.end();
});
}
ipToNumber(ip) { ipToNumber(ip) {
return ip.split('.') return ip.split('.')
.reduce((acc, octet) => (acc << 8) + parseInt(octet), 0) >>> 0; .reduce((acc, octet) => (acc << 8) + parseInt(octet), 0) >>> 0;
@ -100,11 +174,16 @@ class WebSocketScannerWorker {
} }
} }
// Handle messages from the main thread
parentPort.on('message', async (data) => { parentPort.on('message', async (data) => {
if (data.type === 'scan') { if (data.type === 'scan') {
const scanner = new WebSocketScannerWorker(); const scanner = new WebSocketScannerWorker();
const results = await scanner.scanRange(data.startIP, data.endIP, data.port, data.protocol); const results = await scanner.scanRange(
data.startIP,
data.endIP,
data.port,
data.protocol,
data.objectType
);
parentPort.postMessage({ type: 'scanComplete', results }); parentPort.postMessage({ type: 'scanComplete', results });
} }
}); });

View File

@ -27,7 +27,8 @@ export class SocketClient {
this.fileManager = new FileManager(this); this.fileManager = new FileManager(this);
this.printerManager = new PrinterManager(this); this.printerManager = new PrinterManager(this);
this.documentPrinterManager = new DocumentPrinterManager(this); this.documentPrinterManager = new DocumentPrinterManager(this);
this.scanner = new WebSocketScanner({ maxThreads: 50 }); this.scanSessions = new Map();
this.scanSessionTokens = new Map();
this.readLine = null; this.readLine = null;
this.objectEventCallbacks = new Map(); this.objectEventCallbacks = new Map();
sendIPC("setOnline", false); sendIPC("setOnline", false);
@ -707,64 +708,198 @@ export class SocketClient {
//-------------------------------------- RE-WRITE ENDS HERE --------------------------------------- //-------------------------------------- RE-WRITE ENDS HERE ---------------------------------------
emitScanNetworkEvent(eventType, eventData) {
this.objectEvent({
objectType: "host",
_id: this.id,
eventType,
eventData,
});
}
async stopScanSession(sessionId, { emitStopped = false } = {}) {
const scanner = this.scanSessions.get(sessionId);
if (!scanner) {
return false;
}
this.scanSessions.delete(sessionId);
scanner.removeAllListeners("serviceFound");
scanner.removeAllListeners("scanProgress");
scanner.removeAllListeners("scanPassComplete");
scanner.removeAllListeners("scanComplete");
await scanner.stopScan();
if (emitStopped) {
this.emitScanNetworkEvent("scanNetworkComplete", {
sessionId,
success: true,
stopped: true,
});
}
return true;
}
async handleScanNetworkStart(data, callback) { async handleScanNetworkStart(data, callback) {
if (this.scanner.scanning == false) { const protocol = data?.protocol;
try { const sessionId = data?.sessionId;
this.scanner = new WebSocketScanner({ maxThreads: 50 });
// Listen for found services
this.scanner.on("serviceFound", (data) => {
logger.info(
`Found websocket service at ${data.hostname} (${data.ip})`,
);
this.socket.emit("notify_scan_network_found", data);
});
// Listen for scan progress if (!sessionId) {
this.scanner.on("scanProgress", ({ currentIP, progress }) => { if (typeof callback === "function") {
logger.info( callback({ success: false, error: "Scan sessionId is required" });
`Scanning ${currentIP} (${progress.toFixed(2)}% complete)`, }
); return;
this.socket.emit("notify_scan_network_progress", { }
currentIP: currentIP,
progress: progress,
});
});
// Start scanning on port if (protocol === "serial" || protocol === "system") {
if (typeof callback === "function") {
callback({
success: false,
error: "Cannot scan serial or system protocols",
});
}
return;
}
const startToken = {};
this.scanSessionTokens.set(sessionId, startToken);
await this.stopScanSession(sessionId);
if (this.scanSessionTokens.get(sessionId) !== startToken) {
return;
}
try {
const scanner = new WebSocketScanner({ maxThreads: 50 });
this.scanSessions.set(sessionId, scanner);
scanner.on("serviceFound", (found) => {
logger.info( logger.info(
"Scanning network for websocket services on port:", `Found service at ${found.hostname || found.ip} (${found.ip})`,
data?.port || 7125,
"using protocol:",
data?.protocol || "ws",
); );
this.scanner this.emitScanNetworkEvent("scanNetworkFound", {
.scanNetwork(data?.port || 7125, data?.protocol || "ws") ...found,
.then((foundServices) => { sessionId,
logger.info("Scan complete. Found services:", foundServices); });
this.socket.emit("notify_scan_network_complete", foundServices); });
})
.catch((error) => { scanner.on("scanProgress", ({ currentIP, progress }) => {
logger.error("Scan error:", error); logger.debug(
this.socket.emit("notify_scan_network_complete", false); `Scanning ${currentIP} (${progress.toFixed(2)}% complete)`,
);
this.emitScanNetworkEvent("scanNetworkProgress", {
currentIP,
progress,
sessionId,
});
});
scanner.on("scanPassComplete", (foundServices) => {
if (this.scanSessions.get(sessionId) !== scanner) {
return;
}
logger.info("Scan pass complete. Found services:", foundServices);
this.emitScanNetworkEvent("scanNetworkComplete", {
success: true,
devices: foundServices,
sessionId,
looping: true,
});
});
const port =
data?.port || (data?.objectType === "documentPrinter" ? 631 : 7125);
const scanProtocol =
protocol || (data?.objectType === "documentPrinter" ? "ipp" : "ws");
const objectType = data?.objectType || "printer";
logger.info(
"Scanning network on port:",
port,
"using protocol:",
scanProtocol,
"objectType:",
objectType,
"sessionId:",
sessionId,
);
if (typeof callback === "function") {
callback({ success: true, sessionId });
}
scanner
.scanNetwork(port, scanProtocol, objectType, { loop: true })
.then((foundServices) => {
if (this.scanSessions.get(sessionId) !== scanner) {
return;
}
this.scanSessions.delete(sessionId);
logger.info("Scan session ended. Found services:", foundServices);
this.emitScanNetworkEvent("scanNetworkComplete", {
success: true,
devices: foundServices,
sessionId,
looping: false,
}); });
} catch (error) { })
logger.error("Scan error:", error); .catch((error) => {
this.socket.emit("notify_scan_network_complete", false); if (this.scanSessions.get(sessionId) !== scanner) {
return;
}
this.scanSessions.delete(sessionId);
logger.error("Scan error:", error);
this.emitScanNetworkEvent("scanNetworkComplete", {
success: false,
error: error?.message || "Scan failed",
devices: [],
sessionId,
});
});
} catch (error) {
logger.error("Scan error:", error);
await this.stopScanSession(sessionId);
this.emitScanNetworkEvent("scanNetworkComplete", {
success: false,
error: error?.message || "Scan failed",
devices: [],
sessionId,
});
if (typeof callback === "function") {
callback({ success: false, error: error?.message || "Scan failed" });
} }
} }
} }
handleScanNetworkStop(callback) { async handleScanNetworkStop(data, callback) {
if (this.scanner.scanning == true) { const sessionId = data?.sessionId;
logger.info("Stopping network scan");
this.scanner.removeAllListeners("serviceFound"); if (sessionId) {
this.scanner.removeAllListeners("scanProgress"); const stopped = await this.stopScanSession(sessionId, {
this.scanner.removeAllListeners("scanComplete"); emitStopped: true,
this.scanner.stopScan(); });
callback(true); if (typeof callback === "function") {
} else { callback(
logger.info("Scan not in progress"); stopped
callback(false); ? { success: true, sessionId }
: {
success: true,
sessionId,
error: "Scan session not in progress",
},
);
}
return;
}
logger.info("Stopping all network scan sessions");
for (const id of Array.from(this.scanSessions.keys())) {
await this.stopScanSession(id, { emitStopped: true });
}
if (typeof callback === "function") {
callback({ success: true });
} }
} }