snxraven/peardock
1
1
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

test enhancements

Browse Source
This commit is contained in:
Raven Scott
2025-11-24 09:06:29 -05:00
parent 46d359f18a
commit 074d8df3cc
15 changed files with 2354 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/server.js
+304 -116
View File
@@ -8,6 +8,11 @@ import os from "os";
import fs from 'fs';
import dotenv from 'dotenv';
import { spawn } from 'child_process';
import * as validation from './utils/validation.js';
import rateLimiter from './utils/rateLimiter.js';
import * as containerConfig from './utils/containerConfig.js';
import { createErrorResponse, sanitizeErrorMessage } from '../utils/errorHandler.js';
import logger from './utils/logger.js';
// Load environment variables from .env file
dotenv.config();
@@ -40,7 +45,7 @@ if (!keyHex) {
// Convert the keyHex to a Buffer
const topic = Buffer.from(keyHex, 'hex');
console.log(`[INFO] Server started with topic: ${topic.toString('hex')}`);
logger.info(`Server started with topic: ${topic.toString('hex')}`);
// Start listening or further implementation logic here
// Join the swarm with the generated topic
@@ -48,7 +53,7 @@ swarm.join(topic, { server: true, client: false });
// Handle incoming peer connections
swarm.on('connection', (peer) => {
console.log('[INFO] Peer connected');
logger.info('Peer connected', { peerId: peer.remotePublicKey?.toString('hex')?.substring(0, 12) });
connectedPeers.add(peer);
peer.on('data', async (data) => {
@@ -57,6 +62,17 @@ swarm.on('connection', (peer) => {
if (!(parsedData.command === 'stats' && Object.keys(parsedData.args).length === 0)) {
console.log(`[DEBUG] Received data from peer: ${JSON.stringify(parsedData)}`);
}
// Rate limiting check
if (!rateLimiter.isAllowed(peer, parsedData.command)) {
console.warn(`[WARN] Rate limit exceeded for peer: ${parsedData.command}`);
peer.write(JSON.stringify({
error: 'Rate limit exceeded. Please wait before making more requests.',
code: 'RATE_LIMIT_EXCEEDED'
}));
return;
}
let response;
switch (parsedData.command) {
@@ -104,9 +120,26 @@ swarm.on('connection', (peer) => {
console.log(`[INFO] Handling 'dockerCommand' with data: ${parsedData.data}`);
try {
const command = parsedData.data.split(' '); // Split the command into executable and args
// Validate command input
const commandStr = validation.sanitizeString(parsedData.data, 500);
if (!commandStr || !commandStr.startsWith('docker ')) {
throw new Error('Invalid command format');
}
// Additional server-side validation
const dangerousPatterns = ['exec', 'run', 'rm -f', 'prune', 'system prune'];
if (dangerousPatterns.some(pattern => commandStr.includes(pattern))) {
throw new Error('Command not allowed for security reasons');
}
const command = commandStr.split(' '); // Split the command into executable and args
const executable = command[0];
const args = command.slice(1);
// Ensure only docker executable
if (executable !== 'docker') {
throw new Error('Only docker commands are allowed');
}
const childProcess = spawn(executable, args);
@@ -263,23 +296,23 @@ swarm.on('connection', (peer) => {
break;
case 'deployContainer':
console.log('[INFO] Handling "deployContainer" command');
logger.info('Handling deployContainer command');
const args = parsedData.args;
try {
// Validate and sanitize container name
if (!args.containerName || typeof args.containerName !== 'string') {
throw new Error('Invalid or missing container name.');
}
if (!/^[a-zA-Z0-9-_]+$/.test(args.containerName)) {
throw new Error('Container name must be alphanumeric and may include dashes or underscores.');
const containerName = validation.sanitizeString(args.containerName, 63);
if (!containerName || !validation.isValidContainerName(containerName)) {
throw new Error('Invalid or missing container name. Must be alphanumeric with dashes/underscores, 1-63 characters.');
}
args.containerName = containerName;
// Validate and sanitize image
if (!args.image || typeof args.image !== 'string') {
throw new Error('Invalid or missing Docker image.');
const image = validation.sanitizeString(args.image, 255);
if (!image || !validation.isValidImageName(image)) {
throw new Error('Invalid or missing Docker image name.');
}
args.image = image;
// Check if container name already exists
const existingContainers = await docker.listContainers({ all: true });
@@ -288,7 +321,7 @@ swarm.on('connection', (peer) => {
throw new Error(`Container name '${args.containerName}' already exists.`);
}
console.log(`[INFO] Pulling Docker image "${args.image}"`);
logger.info(`Pulling Docker image: ${args.image}`);
// Pull the Docker image
const pullStream = await docker.pull(args.image);
@@ -296,7 +329,7 @@ swarm.on('connection', (peer) => {
docker.modem.followProgress(pullStream, (err) => (err ? reject(err) : resolve()));
});
console.log(`[INFO] Image "${args.image}" pulled successfully`);
logger.info(`Image pulled successfully: ${args.image}`);
// Build container configuration
const containerConfig = {
@@ -319,17 +352,39 @@ swarm.on('connection', (peer) => {
if (args.env && Array.isArray(args.env)) {
containerConfig.Env = args.env
.filter(e => e.name && e.value !== undefined)
.map(e => `${e.name}=${e.value}`);
.map(e => {
const name = validation.sanitizeEnvVarName(e.name);
const value = validation.sanitizeEnvVarValue(e.value);
return name && value !== null ? `${name}=${value}` : null;
})
.filter(e => e !== null);
}
// Labels
if (args.labels && typeof args.labels === 'object') {
containerConfig.Labels = args.labels;
containerConfig.Labels = {};
for (const [key, value] of Object.entries(args.labels)) {
const sanitizedKey = validation.sanitizeLabelKey(key);
const sanitizedValue = validation.sanitizeLabelValue(value);
if (sanitizedKey && sanitizedValue !== null) {
containerConfig.Labels[sanitizedKey] = sanitizedValue;
}
}
}
// Hostname and domainname
if (args.hostname) containerConfig.Hostname = args.hostname;
if (args.domainname) containerConfig.Domainname = args.domainname;
if (args.hostname) {
const hostname = validation.sanitizeString(args.hostname, 253);
if (validation.isValidHostname(hostname)) {
containerConfig.Hostname = hostname;
}
}
if (args.domainname) {
const domainname = validation.sanitizeString(args.domainname, 253);
if (validation.isValidHostname(domainname)) {
containerConfig.Domainname = domainname;
}
}
// User
if (args.user) containerConfig.User = args.user;
@@ -368,21 +423,26 @@ swarm.on('connection', (peer) => {
if (args.ports && Array.isArray(args.ports)) {
hostConfig.PortBindings = {};
args.ports.forEach((portStr) => {
// Support both "host:container/protocol" and "container/protocol" formats
if (portStr.includes(':')) {
const [hostPort, rest] = portStr.split(':');
const [containerPort, protocol] = rest.split('/');
hostConfig.PortBindings[`${containerPort}/${protocol || 'tcp'}`] = [{ HostPort: hostPort }];
} else {
const [containerPort, protocol] = portStr.split('/');
hostConfig.PortBindings[`${containerPort}/${protocol || 'tcp'}`] = [{ HostPort: containerPort }];
const sanitizedPort = validation.sanitizeString(portStr, 50);
if (validation.isValidPortMapping(sanitizedPort)) {
// Support both "host:container/protocol" and "container/protocol" formats
if (sanitizedPort.includes(':')) {
const [hostPort, rest] = sanitizedPort.split(':');
const [containerPort, protocol] = rest.split('/');
hostConfig.PortBindings[`${containerPort}/${protocol || 'tcp'}`] = [{ HostPort: hostPort }];
} else {
const [containerPort, protocol] = sanitizedPort.split('/');
hostConfig.PortBindings[`${containerPort}/${protocol || 'tcp'}`] = [{ HostPort: containerPort }];
}
}
});
}
// Volumes
if (args.volumes && Array.isArray(args.volumes)) {
hostConfig.Binds = args.volumes.filter(v => v && v.includes(':'));
hostConfig.Binds = args.volumes
.map(v => validation.sanitizeString(v, 500))
.filter(v => v && validation.isValidVolumeMount(v));
}
// Tmpfs
@@ -430,7 +490,9 @@ swarm.on('connection', (peer) => {
// DNS
if (args.dns && Array.isArray(args.dns)) {
hostConfig.Dns = args.dns;
hostConfig.Dns = args.dns
.map(dns => validation.sanitizeString(dns, 50))
.filter(dns => validation.isValidDnsServer(dns));
}
// Extra hosts
@@ -514,8 +576,8 @@ swarm.on('connection', (peer) => {
containerConfig.HostConfig = hostConfig;
// Create the container
console.log('[INFO] Creating the container...');
const container = await docker.createContainer(containerConfig);
logger.info('Creating container', { name: args.containerName });
const container = await docker.createContainer(config);
// Connect to custom network if specified
if (args.customNetwork && args.networkMode !== 'container' && args.networkMode !== 'host' && args.networkMode !== 'none') {
@@ -529,10 +591,10 @@ swarm.on('connection', (peer) => {
}
// Start the container
console.log('[INFO] Starting the container...');
logger.info('Starting container', { name: args.containerName });
await container.start();
console.log(`[INFO] Container "${args.containerName}" deployed successfully`);
logger.info('Container deployed successfully', { name: args.containerName, image: args.image });
// Respond with success message
peer.write(
@@ -554,12 +616,10 @@ swarm.on('connection', (peer) => {
}
}
} catch (err) {
console.error(`[ERROR] Failed to deploy container: ${err.message}`);
peer.write(
JSON.stringify({
error: `Failed to deploy container: ${err.message}`,
})
);
logger.error('Failed to deploy container', { error: err.message, containerName: args?.containerName });
const errorResponse = createErrorResponse(err);
errorResponse.error = sanitizeErrorMessage(errorResponse.error);
peer.write(JSON.stringify(errorResponse));
}
break;
@@ -590,18 +650,27 @@ swarm.on('connection', (peer) => {
peer.write(JSON.stringify(response));
}
} catch (err) {
console.error(`[ERROR] Failed to handle data from peer: ${err.message}`);
peer.write(JSON.stringify({ error: err.message }));
logger.error('Failed to handle data from peer', { error: err.message, command: parsedData?.command });
// Sanitize error messages to prevent information leakage
const errorMessage = err.message.includes('ENOENT') || err.message.includes('EACCES')
? 'Operation failed. Please check permissions and try again.'
: err.message.length > 200
? 'An error occurred. Please try again.'
: err.message;
peer.write(JSON.stringify({
error: errorMessage,
code: err.code || 'UNKNOWN_ERROR'
}));
}
});
peer.on('error', (err) => {
console.error(`[ERROR] Peer connection error: ${err.message}`);
logger.error('Peer connection error', { error: err.message });
cleanupPeer(peer);
});
peer.on('close', () => {
console.log('[INFO] Peer disconnected');
logger.info('Peer disconnected');
connectedPeers.delete(peer);
cleanupPeer(peer)
@@ -724,70 +793,90 @@ async function duplicateContainer(name, image, hostname, netmode, cpu, memory, c
// Stream Docker events to all peers
let dockerEventStream = null;
docker.getEvents({}, (err, stream) => {
if (err) {
console.error(`[ERROR] Failed to get Docker events: ${err.message}`);
return;
}
dockerEventStream = stream; // Store reference for cleanup
stream.on('data', async (chunk) => {
try {
const event = JSON.parse(chunk.toString());
if (event.status === "undefined") return
console.log(`[INFO] Docker event received: ${event.status} - ${event.id}`);
// Get updated container list and broadcast it to all connected peers
const containers = await docker.listContainers({ all: true });
const update = { type: 'containers', data: containers };
for (const peer of connectedPeers) {
try {
peer.write(JSON.stringify(update));
} catch (peerErr) {
console.error(`[ERROR] Failed to send update to peer: ${peerErr.message}`);
async function initializeDockerEventStream() {
try {
const stream = await new Promise((resolve, reject) => {
docker.getEvents({}, (err, stream) => {
if (err) {
reject(err);
} else {
resolve(stream);
}
}
} catch (err) {
console.error(`[ERROR] Failed to process Docker event: ${err.message}`);
}
});
});
});
stream.on('error', (err) => {
console.error(`[ERROR] Docker event stream error: ${err.message}`);
});
dockerEventStream = stream; // Store reference for cleanup
stream.on('end', () => {
console.log('[INFO] Docker event stream ended');
dockerEventStream = null;
});
});
stream.on('data', async (chunk) => {
try {
const event = JSON.parse(chunk.toString());
if (event.status === "undefined") return;
logger.info('Docker event received', { status: event.status, id: event.id });
// Collect and stream container stats
docker.listContainers({ all: true }, async (err, containers) => {
if (err) {
console.error(`[ERROR] Failed to list containers for stats: ${err.message}`);
return;
}
// Get updated container list and broadcast it to all connected peers
const containers = await docker.listContainers({ all: true });
const update = { type: 'containers', data: containers };
// Iterate over all containers
containers.forEach((containerInfo) => {
const container = docker.getContainer(containerInfo.Id);
// Use the same logic as listContainers to pre-inspect and extract the IP address
container.inspect((inspectErr, details) => {
let ipAddress = 'No IP Assigned'; // Default fallback
if (!inspectErr && details.NetworkSettings && details.NetworkSettings.Networks) {
const networks = Object.values(details.NetworkSettings.Networks);
if (networks.length > 0 && networks[0].IPAddress) {
ipAddress = networks[0].IPAddress; // Use the first network's IP
for (const peer of connectedPeers) {
try {
peer.write(JSON.stringify(update));
} catch (peerErr) {
logger.error('Failed to send update to peer', { error: peerErr.message });
}
}
} catch (err) {
logger.error('Failed to process Docker event', { error: err.message });
}
});
});
});
stream.on('error', (err) => {
logger.error('Docker event stream error', { error: err.message });
});
stream.on('end', () => {
logger.info('Docker event stream ended');
dockerEventStream = null;
});
} catch (err) {
logger.error('Failed to get Docker events', { error: err.message });
}
}
// Initialize Docker event stream
initializeDockerEventStream();
// Collect and stream container stats (async/await version)
async function initializeContainerStatsCollection() {
try {
const containers = await docker.listContainers({ all: true });
// Iterate over all containers
for (const containerInfo of containers) {
const container = docker.getContainer(containerInfo.Id);
try {
// Use the same logic as listContainers to pre-inspect and extract the IP address
const details = await container.inspect();
let ipAddress = 'No IP Assigned'; // Default fallback
if (details.NetworkSettings && details.NetworkSettings.Networks) {
const networks = Object.values(details.NetworkSettings.Networks);
if (networks.length > 0 && networks[0].IPAddress) {
ipAddress = networks[0].IPAddress; // Use the first network's IP
}
}
} catch (inspectErr) {
logger.debug('Failed to inspect container for IP', { containerId: containerInfo.Id, error: inspectErr.message });
}
}
} catch (err) {
logger.error('Failed to list containers for stats', { error: err.message });
}
}
// Initialize container stats collection
initializeContainerStatsCollection();
// Function to calculate CPU usage percentage
function calculateCPUPercent(stats) {
@@ -1004,40 +1093,139 @@ async function initializeContainerStats(containerInfo) {
return statsData;
}
// Stats cache with TTL
const statsCache = new Map();
const STATS_CACHE_TTL = 1000; // 1 second cache TTL
const STATS_BROADCAST_INTERVAL = 2000; // 2 seconds broadcast interval
// Track container activity for adaptive polling
const containerActivity = new Map(); // containerId -> lastActivity timestamp
/**
* Determine if container is active based on CPU/memory usage
* @param {Object} statsData - Container stats data
* @returns {boolean} - True if container is considered active
*/
function isContainerActive(statsData) {
const cpuThreshold = 1.0; // 1% CPU threshold
const memoryThreshold = 1024 * 1024; // 1MB memory threshold
return statsData.cpu > cpuThreshold || statsData.memory > memoryThreshold;
}
/**
* Update container activity tracking
* @param {string} containerId - Container ID
* @param {Object} statsData - Container stats data
*/
function updateContainerActivity(containerId, statsData) {
if (isContainerActive(statsData)) {
containerActivity.set(containerId, Date.now());
}
}
/**
* Check if stats are cached and still valid
* @param {string} containerId - Container ID
* @returns {Object|null} - Cached stats or null
*/
function getCachedStats(containerId) {
const cached = statsCache.get(containerId);
if (cached && (Date.now() - cached.timestamp) < STATS_CACHE_TTL) {
return cached.data;
}
return null;
}
/**
* Cache stats data
* @param {string} containerId - Container ID
* @param {Object} statsData - Stats data to cache
*/
function cacheStats(containerId, statsData) {
statsCache.set(containerId, {
data: { ...statsData },
timestamp: Date.now()
});
}
async function handleStatsBroadcast() {
const containerStats = {};
let lastBroadcast = Date.now();
// Periodically update stats and broadcast
// Increased interval to 2000ms (2 seconds) for better performance
setInterval(async () => {
try {
await collectContainerStats(containerStats);
// Create clean stats objects without stream references for serialization
const aggregatedStats = Object.values(containerStats).map(statsData => ({
id: statsData.id,
name: statsData.name,
cpu: statsData.cpu,
memory: statsData.memory,
ip: statsData.ip
}));
const now = Date.now();
const timeSinceLastBroadcast = now - lastBroadcast;
// Only broadcast if there are stats to send
if (aggregatedStats.length > 0) {
const response = { type: 'allStats', data: aggregatedStats };
for (const peer of connectedPeers) {
try {
peer.write(JSON.stringify(response));
} catch (err) {
console.error(`[ERROR] Failed to send stats to peer: ${err.message}`);
// Only broadcast if enough time has passed
if (timeSinceLastBroadcast >= STATS_BROADCAST_INTERVAL) {
// Create clean stats objects without stream references for serialization
const aggregatedStats = [];
for (const [containerId, statsData] of Object.entries(containerStats)) {
// Check cache first
const cached = getCachedStats(containerId);
if (cached && !isContainerActive(statsData)) {
// Use cached data for inactive containers
aggregatedStats.push(cached);
continue;
}
// Update activity tracking
updateContainerActivity(containerId, statsData);
// Create stats object
const statsObj = {
id: statsData.id,
name: statsData.name,
cpu: statsData.cpu,
memory: statsData.memory,
ip: statsData.ip
};
// Cache the stats
cacheStats(containerId, statsObj);
aggregatedStats.push(statsObj);
}
// Only broadcast if there are stats to send and peers connected
if (aggregatedStats.length > 0 && connectedPeers.size > 0) {
const response = { type: 'allStats', data: aggregatedStats };
for (const peer of connectedPeers) {
try {
peer.write(JSON.stringify(response));
} catch (err) {
console.error(`[ERROR] Failed to send stats to peer: ${err.message}`);
}
}
lastBroadcast = now;
}
}
// Clean up old cache entries
for (const [containerId, cached] of statsCache.entries()) {
if ((now - cached.timestamp) > STATS_CACHE_TTL * 10) {
statsCache.delete(containerId);
}
}
// Clean up old activity tracking
const activityTimeout = 60000; // 1 minute
for (const [containerId, lastActivity] of containerActivity.entries()) {
if ((now - lastActivity) > activityTimeout) {
containerActivity.delete(containerId);
}
}
} catch (err) {
console.error(`[ERROR] Failed to collect/broadcast stats: ${err.message}`);
}
}, 2000); // Send stats every 2 seconds (reduced frequency for better performance)
}, 1000); // Check every second, but broadcast based on interval
}
// Start the stats broadcast