feat(events): Sanitize all incoming event data before using it

src/lib/__tests__/config-file-schema.test.js

@@ -90,6 +90,25 @@ describe('config-file-schema', () => {
                     udpServerConfig: {
                         serverHost: 'localhost',
                         portUserActivityEvents: 9999,
+                        messageQueue: {
+                            maxConcurrent: 10,
+                            maxSize: 200,
+                            dropStrategy: 'oldest',
+                            backpressureThreshold: 80,
+                        },
+                        rateLimit: {
+                            enable: false,
+                            maxMessagesPerMinute: 600,
+                        },
+                        maxMessageSize: 65507,
+                        queueMetrics: {
+                            influxdb: {
+                                enable: false,
+                                writeFrequency: 20000,
+                                measurementName: 'user_events_queue',
+                                tags: [],
+                            },
+                        },
                     },
                     tags: null,
                 },
@@ -106,6 +125,25 @@ describe('config-file-schema', () => {
                     udpServerConfig: {
                         serverHost: 'localhost',
                         portLogEvents: 9998,
+                        messageQueue: {
+                            maxConcurrent: 10,
+                            maxSize: 200,
+                            dropStrategy: 'oldest',
+                            backpressureThreshold: 80,
+                        },
+                        rateLimit: {
+                            enable: false,
+                            maxMessagesPerMinute: 600,
+                        },
+                        maxMessageSize: 65507,
+                        queueMetrics: {
+                            influxdb: {
+                                enable: false,
+                                writeFrequency: 20000,
+                                measurementName: 'log_events_queue',
+                                tags: [],
+                            },
+                        },
                     },
                     source: {
                         engine: {

src/lib/__tests__/udp-queue-manager.test.js

@@ -0,0 +1,415 @@
+/**
+ * Tests for UDP Queue Manager
+ */
+
+import { describe, it, expect, beforeEach, jest, afterEach } from '@jest/globals';
+import { UdpQueueManager, sanitizeField } from '../udp-queue-manager.js';
+
+describe('sanitizeField', () => {
+    it('should remove control characters from string', () => {
+        const input = 'Hello\x00World\x1FTest\x7F';
+        const result = sanitizeField(input);
+        expect(result).toBe('HelloWorldTest');
+    });
+
+    it('should limit string length to default 500 characters', () => {
+        const input = 'a'.repeat(1000);
+        const result = sanitizeField(input);
+        expect(result).toHaveLength(500);
+    });
+
+    it('should limit string length to custom maxLength', () => {
+        const input = 'a'.repeat(1000);
+        const result = sanitizeField(input, 100);
+        expect(result).toHaveLength(100);
+    });
+
+    it('should handle non-string input by converting to string', () => {
+        const result = sanitizeField(12345);
+        expect(result).toBe('12345');
+    });
+
+    it('should handle empty string', () => {
+        const result = sanitizeField('');
+        expect(result).toBe('');
+    });
+
+    it('should remove newlines and carriage returns', () => {
+        const input = 'Line1\nLine2\rLine3';
+        const result = sanitizeField(input);
+        expect(result).toBe('Line1Line2Line3');
+    });
+
+    it('should preserve normal characters', () => {
+        const input = 'Hello World! 123 @#$%';
+        const result = sanitizeField(input);
+        expect(result).toBe('Hello World! 123 @#$%');
+    });
+});
+
+describe('UdpQueueManager', () => {
+    let queueManager;
+    let mockLogger;
+    let config;
+
+    beforeEach(() => {
+        mockLogger = {
+            warn: jest.fn(),
+            info: jest.fn(),
+            error: jest.fn(),
+            debug: jest.fn(),
+        };
+
+        config = {
+            messageQueue: {
+                maxConcurrent: 5,
+                maxSize: 10,
+                dropStrategy: 'oldest',
+                backpressureThreshold: 80,
+            },
+            rateLimit: {
+                enable: false,
+                maxMessagesPerMinute: 60,
+            },
+            maxMessageSize: 1024,
+        };
+
+        queueManager = new UdpQueueManager(config, mockLogger, 'test-queue');
+    });
+
+    afterEach(() => {
+        jest.clearAllMocks();
+    });
+
+    describe('constructor', () => {
+        it('should initialize with correct config', () => {
+            expect(queueManager.config).toEqual(config);
+            expect(queueManager.logger).toEqual(mockLogger);
+            expect(queueManager.queueType).toBe('test-queue');
+        });
+
+        it('should initialize rate limiter when enabled', () => {
+            const configWithRateLimit = {
+                ...config,
+                rateLimit: { enable: true, maxMessagesPerMinute: 60 },
+            };
+            const qm = new UdpQueueManager(configWithRateLimit, mockLogger, 'test');
+            expect(qm.rateLimiter).toBeTruthy();
+        });
+
+        it('should not initialize rate limiter when disabled', () => {
+            expect(queueManager.rateLimiter).toBeNull();
+        });
+    });
+
+    describe('validateMessageSize', () => {
+        it('should accept message within size limit', () => {
+            const message = Buffer.from('small message');
+            expect(queueManager.validateMessageSize(message)).toBe(true);
+        });
+
+        it('should reject message exceeding size limit', () => {
+            const message = Buffer.alloc(2000);
+            expect(queueManager.validateMessageSize(message)).toBe(false);
+        });
+
+        it('should handle string messages', () => {
+            const message = 'test string';
+            expect(queueManager.validateMessageSize(message)).toBe(true);
+        });
+    });
+
+    describe('checkRateLimit', () => {
+        it('should return true when rate limiting is disabled', () => {
+            expect(queueManager.checkRateLimit()).toBe(true);
+        });
+
+        it('should respect rate limit when enabled', () => {
+            const configWithRateLimit = {
+                ...config,
+                rateLimit: { enable: true, maxMessagesPerMinute: 5 },
+            };
+            const qm = new UdpQueueManager(configWithRateLimit, mockLogger, 'test');
+
+            // Should accept first 5 messages
+            for (let i = 0; i < 5; i++) {
+                expect(qm.checkRateLimit()).toBe(true);
+            }
+
+            // Should reject 6th message
+            expect(qm.checkRateLimit()).toBe(false);
+        });
+
+        it('should reset rate limit after 1 minute', async () => {
+            const configWithRateLimit = {
+                ...config,
+                rateLimit: { enable: true, maxMessagesPerMinute: 2 },
+            };
+            const qm = new UdpQueueManager(configWithRateLimit, mockLogger, 'test');
+
+            // Fill up the rate limit
+            expect(qm.checkRateLimit()).toBe(true);
+            expect(qm.checkRateLimit()).toBe(true);
+            expect(qm.checkRateLimit()).toBe(false);
+
+            // Fast-forward time by 61 seconds
+            jest.useFakeTimers();
+            jest.advanceTimersByTime(61000);
+
+            // Should accept messages again
+            expect(qm.checkRateLimit()).toBe(true);
+
+            jest.useRealTimers();
+        });
+    });
+
+    describe('addToQueue', () => {
+        it('should queue and process messages', async () => {
+            const processFunction = jest.fn().mockResolvedValue();
+            const result = await queueManager.addToQueue(processFunction);
+
+            expect(result).toBe(true);
+
+            // Wait for queue to process
+            await queueManager.queue.onIdle();
+
+            expect(processFunction).toHaveBeenCalled();
+        });
+
+        it('should reject messages when queue is full', async () => {
+            // Use very slow processing to ensure queue fills up
+            const processFunction = jest.fn().mockImplementation(
+                () =>
+                    new Promise((resolve) => {
+                        // Never resolve during test - keep queue full
+                    })
+            );
+
+            // Rapidly add more messages than queue can hold
+            // maxConcurrent: 5, maxSize: 10
+            // Queue.size only counts pending (not currently processing)
+            // So: 5 processing + 5 pending (queue.size=5) = 10 total capacity
+            // When we try to add 20, some should be rejected
+            const promises = [];
+            for (let i = 0; i < 20; i++) {
+                promises.push(queueManager.addToQueue(processFunction));
+            }
+
+            // Wait for all attempts to complete
+            const results = await Promise.all(promises);
+
+            // Count rejections and acceptances
+            const rejectedCount = results.filter((r) => r === false).length;
+            const acceptedCount = results.filter((r) => r === true).length;
+
+            // We should have some rejections (at least a few)
+            expect(rejectedCount).toBeGreaterThanOrEqual(5);
+            // And total should be 20
+            expect(acceptedCount + rejectedCount).toBe(20);
+        });
+
+        it('should track metrics for processed messages', async () => {
+            const processFunction = jest.fn().mockResolvedValue();
+            await queueManager.addToQueue(processFunction);
+
+            await queueManager.queue.onIdle();
+
+            const metrics = await queueManager.getMetrics();
+            expect(metrics.messagesReceived).toBe(1);
+            expect(metrics.messagesQueued).toBe(1);
+            expect(metrics.messagesProcessed).toBe(1);
+        });
+
+        it('should track failed messages', async () => {
+            const processFunction = jest.fn().mockRejectedValue(new Error('Test error'));
+            await queueManager.addToQueue(processFunction);
+
+            await queueManager.queue.onIdle();
+
+            const metrics = await queueManager.getMetrics();
+            expect(metrics.messagesFailed).toBe(1);
+        });
+    });
+
+    describe('handleRateLimitDrop', () => {
+        it('should increment rate limit drop counter', async () => {
+            await queueManager.handleRateLimitDrop();
+
+            const metrics = await queueManager.getMetrics();
+            expect(metrics.messagesDroppedRateLimit).toBe(1);
+            expect(metrics.messagesDroppedTotal).toBe(1);
+        });
+    });
+
+    describe('handleSizeDrop', () => {
+        it('should increment size drop counter', async () => {
+            await queueManager.handleSizeDrop();
+
+            const metrics = await queueManager.getMetrics();
+            expect(metrics.messagesDroppedSize).toBe(1);
+            expect(metrics.messagesDroppedTotal).toBe(1);
+        });
+    });
+
+    describe('getMetrics', () => {
+        it('should return all metrics', async () => {
+            const metrics = await queueManager.getMetrics();
+
+            expect(metrics).toHaveProperty('queueSize');
+            expect(metrics).toHaveProperty('queueMaxSize');
+            expect(metrics).toHaveProperty('queueUtilizationPct');
+            expect(metrics).toHaveProperty('messagesReceived');
+            expect(metrics).toHaveProperty('messagesQueued');
+            expect(metrics).toHaveProperty('messagesProcessed');
+            expect(metrics).toHaveProperty('messagesFailed');
+            expect(metrics).toHaveProperty('messagesDroppedTotal');
+            expect(metrics).toHaveProperty('messagesDroppedRateLimit');
+            expect(metrics).toHaveProperty('messagesDroppedQueueFull');
+            expect(metrics).toHaveProperty('messagesDroppedSize');
+            expect(metrics).toHaveProperty('processingTimeAvgMs');
+            expect(metrics).toHaveProperty('processingTimeP95Ms');
+            expect(metrics).toHaveProperty('processingTimeMaxMs');
+            expect(metrics).toHaveProperty('backpressureActive');
+        });
+
+        it('should calculate queue utilization correctly', async () => {
+            const processFunction = jest.fn().mockImplementation(
+                () =>
+                    new Promise((resolve) => {
+                        // Never resolve - keep in pending state
+                    })
+            );
+
+            // Add messages that will be pending
+            for (let i = 0; i < 5; i++) {
+                await queueManager.addToQueue(processFunction);
+            }
+
+            // Check metrics - pending count should be > 0
+            const metrics = await queueManager.getMetrics();
+            expect(metrics.queuePending).toBeGreaterThan(0);
+            expect(metrics.messagesQueued).toBe(5);
+        });
+    });
+
+    describe('clearMetrics', () => {
+        it('should reset all metrics', async () => {
+            // Generate some metrics
+            const processFunction = jest.fn().mockResolvedValue();
+            await queueManager.addToQueue(processFunction);
+            await queueManager.handleRateLimitDrop();
+            await queueManager.handleSizeDrop();
+
+            await queueManager.queue.onIdle();
+
+            // Clear metrics
+            await queueManager.clearMetrics();
+
+            const metrics = await queueManager.getMetrics();
+            expect(metrics.messagesReceived).toBe(0);
+            expect(metrics.messagesQueued).toBe(0);
+            expect(metrics.messagesProcessed).toBe(0);
+            expect(metrics.messagesFailed).toBe(0);
+            expect(metrics.messagesDroppedTotal).toBe(0);
+            expect(metrics.messagesDroppedRateLimit).toBe(0);
+            expect(metrics.messagesDroppedQueueFull).toBe(0);
+            expect(metrics.messagesDroppedSize).toBe(0);
+        });
+    });
+
+    describe('checkBackpressure', () => {
+        it('should activate backpressure when threshold exceeded', async () => {
+            const processFunction = jest.fn().mockImplementation(
+                () =>
+                    new Promise((resolve) => {
+                        // Never resolve - keep queue full
+                    })
+            );
+
+            // Fill queue beyond backpressure threshold (80% of maxSize=10 means 8)
+            // Queue.size only counts pending items (not currently processing)
+            // So to get queue.size = 8, we need: 5 processing + 8 pending = 13 total
+            const promises = [];
+            for (let i = 0; i < 13; i++) {
+                promises.push(queueManager.addToQueue(processFunction));
+            }
+            await Promise.all(promises);
+
+            // Check backpressure
+            await queueManager.checkBackpressure();
+
+            expect(mockLogger.warn).toHaveBeenCalledWith(
+                expect.stringContaining('Backpressure detected')
+            );
+        });
+
+        it('should clear backpressure when utilization drops', async () => {
+            queueManager.backpressureActive = true;
+
+            await queueManager.checkBackpressure();
+
+            expect(mockLogger.info).toHaveBeenCalledWith(
+                expect.stringContaining('Backpressure cleared')
+            );
+        });
+    });
+
+    describe('CircularBuffer', () => {
+        it('should track processing times', async () => {
+            const processFunction = jest.fn().mockImplementation(
+                () =>
+                    new Promise((resolve) => {
+                        setTimeout(resolve, 10);
+                    })
+            );
+
+            await queueManager.addToQueue(processFunction);
+            await queueManager.queue.onIdle();
+
+            const metrics = await queueManager.getMetrics();
+            expect(metrics.processingTimeAvgMs).toBeGreaterThan(0);
+            expect(metrics.processingTimeMaxMs).toBeGreaterThan(0);
+        });
+
+        it('should calculate 95th percentile', async () => {
+            // Add messages with varying processing times
+            for (let i = 0; i < 20; i++) {
+                const processFunction = jest.fn().mockImplementation(
+                    () =>
+                        new Promise((resolve) => {
+                            setTimeout(resolve, i * 5);
+                        })
+                );
+                await queueManager.addToQueue(processFunction);
+            }
+
+            await queueManager.queue.onIdle();
+
+            const metrics = await queueManager.getMetrics();
+            expect(metrics.processingTimeP95Ms).toBeGreaterThan(0);
+        });
+    });
+
+    describe('logDroppedMessages', () => {
+        it('should log dropped messages after 60 seconds', async () => {
+            jest.useFakeTimers();
+
+            queueManager.droppedSinceLastLog = 5;
+
+            jest.advanceTimersByTime(61000);
+
+            queueManager.logDroppedMessages();
+
+            expect(mockLogger.warn).toHaveBeenCalledWith(
+                expect.stringContaining('Dropped 5 messages')
+            );
+
+            jest.useRealTimers();
+        });
+
+        it('should not log if no messages dropped', () => {
+            queueManager.logDroppedMessages();
+            expect(mockLogger.warn).not.toHaveBeenCalled();
+        });
+    });
+});

src/lib/udp-queue-manager.js

@@ -162,6 +162,23 @@ class RateLimiter {
     }
 }
 
+/**
+ * Sanitize input field by removing control characters and limiting length
+ *
+ * @param {string} field - Field to sanitize
+ * @param {number} maxLength - Maximum length (default: 500)
+ * @returns {string} Sanitized field
+ */
+export function sanitizeField(field, maxLength = 500) {
+    if (typeof field !== 'string') {
+        return String(field).slice(0, maxLength);
+    }
+
+    return field
+        .replace(/[\x00-\x1F\x7F]/g, '') // Remove control characters
+        .slice(0, maxLength);
+}
+
 /**
  * UDP Queue Manager
  * Manages message queue, rate limiting, metrics tracking, and input validation
@@ -225,23 +242,6 @@ export class UdpQueueManager {
         this.lastDropLog = Date.now();
     }
 
-    /**
-     * Sanitize input field by removing control characters and limiting length
-     *
-     * @param {string} field - Field to sanitize
-     * @param {number} maxLength - Maximum length (default: 500)
-     * @returns {string} Sanitized field
-     */
-    sanitizeField(field, maxLength = 500) {
-        if (typeof field !== 'string') {
-            return String(field).slice(0, maxLength);
-        }
-
-        return field
-            .replace(/[\x00-\x1F\x7F]/g, '') // Remove control characters
-            .slice(0, maxLength);
-    }
-
     /**
      * Validate message size
      *

src/lib/udp_handlers/log_events/__tests__/sanitization.test.js

@@ -0,0 +1,329 @@
+/**
+ * Tests for input sanitization in log event handlers
+ */
+
+import { describe, it, expect, beforeEach, jest } from '@jest/globals';
+import { sanitizeField } from '../../../udp-queue-manager.js';
+
+// Mock globals before importing handlers
+jest.unstable_mockModule('../../../../globals.js', () => ({
+    default: {
+        logger: {
+            verbose: jest.fn(),
+            debug: jest.fn(),
+            warn: jest.fn(),
+            error: jest.fn(),
+        },
+        config: {
+            get: jest.fn().mockReturnValue(false), // Disable performance monitor by default
+        },
+        appNames: [],
+    },
+}));
+
+// Import handlers after mocking
+const { processEngineEvent } = await import('../handlers/engine-handler.js');
+const { processProxyEvent } = await import('../handlers/proxy-handler.js');
+const { processRepositoryEvent } = await import('../handlers/repository-handler.js');
+const { processSchedulerEvent } = await import('../handlers/scheduler-handler.js');
+const { processQixPerfEvent } = await import('../handlers/qix-perf-handler.js');
+
+describe('Log Event Handler Sanitization', () => {
+    beforeEach(() => {
+        jest.clearAllMocks();
+    });
+    describe('Engine Event Handler', () => {
+        it('should sanitize control characters in message field', () => {
+            const msg = [
+                '/qseow-engine/',
+                '1',
+                '2021-11-09T15:37:26.028+0200',
+                '2021-11-09 15:37:26,028',
+                'ERROR',
+                'hostname.example.com',
+                'System.Engine',
+                'DOMAIN\\user',
+                'Test message with\x00control\x1Fcharacters\x7F', // Field 8: message
+                '550e8400-e29b-41d4-a716-446655440000',
+                'INTERNAL',
+                'sa_scheduler',
+                '2021-11-09T15:37:26.028+0200',
+                '550e8400-e29b-41d4-a716-446655440000',
+                '12.345.0',
+                '2021-11-09T15:37:26.028+0200',
+                'Traffic',
+                '550e8400-e29b-41d4-a716-446655440000',
+                '550e8400-e29b-41d4-a716-446655440000',
+            ];
+
+            const result = processEngineEvent(msg);
+            expect(result.message).not.toContain('\x00');
+            expect(result.message).not.toContain('\x1F');
+            expect(result.message).not.toContain('\x7F');
+            expect(result.message).toBe('Test message withcontrolcharacters');
+        });
+
+        it('should limit message length to 1000 characters', () => {
+            const longMessage = 'x'.repeat(2000);
+            const msg = [
+                '/qseow-engine/',
+                '1',
+                '2021-11-09T15:37:26.028+0200',
+                '2021-11-09 15:37:26,028',
+                'ERROR',
+                'hostname.example.com',
+                'System.Engine',
+                'DOMAIN\\user',
+                longMessage,
+                '550e8400-e29b-41d4-a716-446655440000',
+                'INTERNAL',
+                'sa_scheduler',
+                '2021-11-09T15:37:26.028+0200',
+                '550e8400-e29b-41d4-a716-446655440000',
+                '12.345.0',
+                '2021-11-09T15:37:26.028+0200',
+                'Traffic',
+                '550e8400-e29b-41d4-a716-446655440000',
+                '550e8400-e29b-41d4-a716-446655440000',
+            ];
+
+            const result = processEngineEvent(msg);
+            expect(result.message).toHaveLength(1000);
+        });
+
+        it('should sanitize all string fields', () => {
+            const msg = [
+                '/qseow-engine/\x00',
+                '1',
+                '2021-11-09T15:37:26.028+0200',
+                '2021-11-09 15:37:26,028',
+                'ERROR\x01',
+                'hostname\x02.example.com',
+                'System.Engine\x03',
+                'DOMAIN\\user\x04',
+                'Message\x05',
+                '550e8400-e29b-41d4-a716-446655440000',
+                'INTERNAL\x06',
+                'sa_scheduler\x07',
+                '2021-11-09T15:37:26.028+0200',
+                '550e8400-e29b-41d4-a716-446655440000',
+                '12.345.0\x08',
+                '2021-11-09T15:37:26.028+0200',
+                'Traffic\x09',
+                '550e8400-e29b-41d4-a716-446655440000',
+                '550e8400-e29b-41d4-a716-446655440000',
+            ];
+
+            const result = processEngineEvent(msg);
+
+            // Check no control characters remain
+            expect(result.source).not.toMatch(/[\x00-\x1F\x7F]/);
+            expect(result.level).not.toMatch(/[\x00-\x1F\x7F]/);
+            expect(result.host).not.toMatch(/[\x00-\x1F\x7F]/);
+            expect(result.subsystem).not.toMatch(/[\x00-\x1F\x7F]/);
+            expect(result.windows_user).not.toMatch(/[\x00-\x1F\x7F]/);
+            expect(result.message).not.toMatch(/[\x00-\x1F\x7F]/);
+            expect(result.user_directory).not.toMatch(/[\x00-\x1F\x7F]/);
+            expect(result.user_id).not.toMatch(/[\x00-\x1F\x7F]/);
+            expect(result.engine_exe_version).not.toMatch(/[\x00-\x1F\x7F]/);
+            expect(result.entry_type).not.toMatch(/[\x00-\x1F\x7F]/);
+        });
+    });
+
+    describe('Proxy Event Handler', () => {
+        it('should sanitize exception message field', () => {
+            const msg = [
+                '/qseow-proxy/',
+                '1',
+                '2021-11-09T15:37:26.028+0200',
+                '2021-11-09 15:37:26,028',
+                'ERROR',
+                'hostname.example.com',
+                'Service.Proxy',
+                'DOMAIN\\user',
+                'Test message',
+                'Exception:\x00Test\x1FError\x7F', // Field 9: exception_message
+                'INTERNAL',
+                'sa_scheduler',
+                'TestCommand',
+                '500',
+                'Origin',
+                '/context',
+            ];
+
+            const result = processProxyEvent(msg);
+            expect(result.exception_message).not.toContain('\x00');
+            expect(result.exception_message).not.toContain('\x1F');
+            expect(result.exception_message).not.toContain('\x7F');
+        });
+    });
+
+    describe('Repository Event Handler', () => {
+        it('should sanitize command and result_code fields', () => {
+            const msg = [
+                '/qseow-repository/',
+                '1',
+                '2021-11-09T15:37:26.028+0200',
+                '2021-11-09 15:37:26,028',
+                'WARN',
+                'hostname.example.com',
+                'Service.Repository',
+                'DOMAIN\\user',
+                'Test message',
+                'Exception message',
+                'INTERNAL',
+                'sa_scheduler',
+                'Check\x00service\x01status', // Field 12: command
+                '500\x02', // Field 13: result_code
+                'Origin',
+                '/context',
+            ];
+
+            const result = processRepositoryEvent(msg);
+            expect(result.command).not.toMatch(/[\x00-\x1F\x7F]/);
+            expect(result.result_code).not.toMatch(/[\x00-\x1F\x7F]/);
+        });
+    });
+
+    describe('Scheduler Event Handler', () => {
+        it('should sanitize task and app names', () => {
+            const msg = [
+                '/qseow-scheduler/',
+                '1',
+                '2021-11-09T19:37:44.331+0100',
+                '2021-11-09 19:37:44,331',
+                'ERROR',
+                'hostname.example.com',
+                'System.Scheduler',
+                'DOMAIN\\user',
+                'Reload failed',
+                'Exception',
+                'LAB',
+                'goran',
+                'LAB\\goran',
+                'Task\x00Name\x01Test', // Field 13: task_name
+                'App\x02Name\x03Test', // Field 14: app_name
+                'dec2a02a-1680-44ef-8dc2-e2bfb180af87',
+                'e7af59a0-c243-480d-9571-08727551a66f',
+                '4831c6a5-34f6-45bb-9d40-73a6e6992670',
+            ];
+
+            const result = processSchedulerEvent(msg);
+            expect(result.task_name).not.toMatch(/[\x00-\x1F\x7F]/);
+            expect(result.app_name).not.toMatch(/[\x00-\x1F\x7F]/);
+            expect(result.task_name).toBe('TaskNameTest');
+            expect(result.app_name).toBe('AppNameTest');
+        });
+    });
+
+    describe('QIX Performance Event Handler', () => {
+        it('should sanitize method and object_type fields', () => {
+            const msg = [
+                '/qseow-qix-perf/',
+                '1',
+                '2021-11-09T19:37:44.331+0100',
+                '2021-11-09 19:37:44,331',
+                'INFO',
+                'hostname.example.com',
+                'System.Engine',
+                'DOMAIN\\user',
+                '550e8400-e29b-41d4-a716-446655440000',
+                'LAB',
+                'goran',
+                '2021-11-09T19:37:44.331+01:00',
+                '550e8400-e29b-41d4-a716-446655440000',
+                '550e8400-e29b-41d4-a716-446655440000',
+                '123',
+                'Global::\x00OpenApp\x01', // Field 15: method
+                '100',
+                '90',
+                '5',
+                '3',
+                '2',
+                '1',
+                'objId123',
+                '1024000',
+                '2048000',
+                'linechart\x02', // Field 25: object_type
+            ];
+
+            const result = processQixPerfEvent(msg);
+            if (result) {
+                expect(result.method).not.toMatch(/[\x00-\x1F\x7F]/);
+                expect(result.object_type).not.toMatch(/[\x00-\x1F\x7F]/);
+            }
+        });
+    });
+
+    describe('sanitizeField edge cases', () => {
+        it('should handle null values', () => {
+            const result = sanitizeField(null);
+            expect(result).toBe('null');
+        });
+
+        it('should handle undefined values', () => {
+            const result = sanitizeField(undefined);
+            expect(result).toBe('undefined');
+        });
+
+        it('should handle numbers', () => {
+            const result = sanitizeField(12345);
+            expect(result).toBe('12345');
+        });
+
+        it('should handle objects by converting to string', () => {
+            const result = sanitizeField({ test: 'value' });
+            expect(result).toContain('[object Object]');
+        });
+
+        it('should handle arrays by converting to string', () => {
+            const result = sanitizeField([1, 2, 3]);
+            expect(result).toBe('1,2,3');
+        });
+
+        it('should remove tab characters', () => {
+            const result = sanitizeField('text\twith\ttabs');
+            expect(result).toBe('textwithtabs');
+        });
+
+        it('should remove all ASCII control characters', () => {
+            // Test characters from 0x00 to 0x1F and 0x7F
+            let input = '';
+            for (let i = 0; i <= 0x1f; i++) {
+                input += String.fromCharCode(i);
+            }
+            input += String.fromCharCode(0x7f);
+            input += 'ValidText';
+
+            const result = sanitizeField(input);
+            expect(result).toBe('ValidText');
+        });
+
+        it('should preserve unicode characters', () => {
+            const result = sanitizeField('Hello 世界 🌍 Привет');
+            expect(result).toBe('Hello 世界 🌍 Привет');
+        });
+
+        it('should handle very long strings efficiently', () => {
+            const longString = 'a'.repeat(10000);
+            const result = sanitizeField(longString, 500);
+            expect(result).toHaveLength(500);
+        });
+    });
+
+    describe('Field length limits', () => {
+        it('should respect different max lengths for different fields', () => {
+            // Source: 100 chars
+            expect(sanitizeField('x'.repeat(200), 100)).toHaveLength(100);
+
+            // Message: 1000 chars
+            expect(sanitizeField('x'.repeat(2000), 1000)).toHaveLength(1000);
+
+            // Subsystem: 200 chars
+            expect(sanitizeField('x'.repeat(300), 200)).toHaveLength(200);
+
+            // Level: 20 chars
+            expect(sanitizeField('x'.repeat(50), 20)).toHaveLength(20);
+        });
+    });
+});

src/lib/udp_handlers/log_events/handlers/engine-handler.js

@@ -4,6 +4,7 @@
 
 import globals from '../../../../globals.js';
 import { isoDateRegex, uuidRegex, formatUserFields } from '../utils/common-utils.js';
+import { sanitizeField } from '../../../udp-queue-manager.js';
 
 /**
  * Process engine log events
@@ -54,26 +55,26 @@ export function processEngineEvent(msg) {
     // session_id: uuid
     // app_id: uuid
     const msgObj = {
-        source: msg[0],
+        source: sanitizeField(msg[0], 100),
         log_row:
             Number.isInteger(parseInt(msg[1], 10)) && parseInt(msg[1], 10) > 0
                 ? parseInt(msg[1], 10)
                 : -1,
-        ts_iso: isoDateRegex.test(msg[2]) ? msg[2] : '',
-        ts_local: isoDateRegex.test(msg[3]) ? msg[3] : '',
-        level: msg[4],
-        host: msg[5],
-        subsystem: msg[6],
-        windows_user: msg[7],
-        message: msg[8],
+        ts_iso: isoDateRegex.test(msg[2]) ? sanitizeField(msg[2], 50) : '',
+        ts_local: isoDateRegex.test(msg[3]) ? sanitizeField(msg[3], 50) : '',
+        level: sanitizeField(msg[4], 20),
+        host: sanitizeField(msg[5], 100),
+        subsystem: sanitizeField(msg[6], 200),
+        windows_user: sanitizeField(msg[7], 100),
+        message: sanitizeField(msg[8], 1000),
         proxy_session_id: uuidRegex.test(msg[9]) ? msg[9] : '',
-        user_directory: msg[10],
-        user_id: msg[11],
-        engine_ts: msg[12],
+        user_directory: sanitizeField(msg[10], 100),
+        user_id: sanitizeField(msg[11], 100),
+        engine_ts: sanitizeField(msg[12], 50),
         process_id: uuidRegex.test(msg[13]) ? msg[13] : '',
-        engine_exe_version: msg[14],
-        server_started: msg[15],
-        entry_type: msg[16],
+        engine_exe_version: sanitizeField(msg[14], 50),
+        server_started: sanitizeField(msg[15], 50),
+        entry_type: sanitizeField(msg[16], 50),
         session_id: uuidRegex.test(msg[17]) ? msg[17] : '',
         app_id: uuidRegex.test(msg[18]) ? msg[18] : '',
     };

src/lib/udp_handlers/log_events/handlers/proxy-handler.js

@@ -4,6 +4,7 @@
 
 import globals from '../../../../globals.js';
 import { isoDateRegex, formatUserFields } from '../utils/common-utils.js';
+import { sanitizeField } from '../../../udp-queue-manager.js';
 
 /**
  * Process proxy log events
@@ -33,22 +34,22 @@ export function processProxyEvent(msg) {
     globals.logger.verbose(`LOG EVENT: ${msg[0]}:${msg[5]}:${msg[4]}, ${msg[6]}, Msg: ${msg[8]}`);
 
     const msgObj = {
-        source: msg[0],
+        source: sanitizeField(msg[0], 100),
         log_row: Number.isInteger(parseInt(msg[1], 10)) ? parseInt(msg[1], 10) : -1,
-        ts_iso: isoDateRegex.test(msg[2]) ? msg[2] : '',
-        ts_local: isoDateRegex.test(msg[3]) ? msg[3] : '',
-        level: msg[4],
-        host: msg[5],
-        subsystem: msg[6],
-        windows_user: msg[7],
-        message: msg[8],
-        exception_message: msg[9],
-        user_directory: msg[10],
-        user_id: msg[11],
-        command: msg[12],
-        result_code: msg[13],
-        origin: msg[14],
-        context: msg[15],
+        ts_iso: isoDateRegex.test(msg[2]) ? sanitizeField(msg[2], 50) : '',
+        ts_local: isoDateRegex.test(msg[3]) ? sanitizeField(msg[3], 50) : '',
+        level: sanitizeField(msg[4], 20),
+        host: sanitizeField(msg[5], 100),
+        subsystem: sanitizeField(msg[6], 200),
+        windows_user: sanitizeField(msg[7], 100),
+        message: sanitizeField(msg[8], 1000),
+        exception_message: sanitizeField(msg[9], 1000),
+        user_directory: sanitizeField(msg[10], 100),
+        user_id: sanitizeField(msg[11], 100),
+        command: sanitizeField(msg[12], 200),
+        result_code: sanitizeField(msg[13], 50),
+        origin: sanitizeField(msg[14], 200),
+        context: sanitizeField(msg[15], 200),
     };
 
     formatUserFields(msgObj);

src/lib/udp_handlers/log_events/handlers/qix-perf-handler.js

@@ -5,6 +5,7 @@
 import globals from '../../../../globals.js';
 import { uuidRegex, formatUserFields } from '../utils/common-utils.js';
 import { processAppSpecificFilters, processAllAppsFilters } from '../filters/qix-perf-filters.js';
+import { sanitizeField } from '../../../udp-queue-manager.js';
 
 /**
  * Process QIX performance log events
@@ -151,23 +152,23 @@ export function processQixPerfEvent(msg) {
     // Event matches filters in the configuration. Continue.
     // Build the event object
     const msgObj = {
-        source: msg[0],
+        source: sanitizeField(msg[0], 100),
         log_row: Number.isInteger(parseInt(msg[1], 10)) ? parseInt(msg[1], 10) : -1,
-        ts_iso: msg[2],
-        ts_local: msg[3],
-        level: msg[4],
-        host: msg[5],
-        subsystem: msg[6],
-        windows_user: msg[7],
+        ts_iso: sanitizeField(msg[2], 50),
+        ts_local: sanitizeField(msg[3], 50),
+        level: sanitizeField(msg[4], 20),
+        host: sanitizeField(msg[5], 100),
+        subsystem: sanitizeField(msg[6], 200),
+        windows_user: sanitizeField(msg[7], 100),
         proxy_session_id: uuidRegex.test(msg[8]) ? msg[8] : '',
-        user_directory: msg[9],
-        user_id: msg[10],
-        engine_ts: msg[11],
+        user_directory: sanitizeField(msg[9], 100),
+        user_id: sanitizeField(msg[10], 100),
+        engine_ts: sanitizeField(msg[11], 50),
         session_id: uuidRegex.test(msg[12]) ? msg[12] : '',
         app_id: uuidRegex.test(msg[13]) ? msg[13] : '',
-        app_name: eventAppName,
-        request_id: msg[14], // Request ID is an integer >= 0, set to -99 otherwise
-        method: msg[15],
+        app_name: sanitizeField(eventAppName, 200),
+        request_id: sanitizeField(msg[14], 50), // Request ID is an integer >= 0, set to -99 otherwise
+        method: sanitizeField(msg[15], 100),
         // Processtime in float milliseconds
         process_time: parseFloat(msg[16]),
         work_time: parseFloat(msg[17]),
@@ -176,7 +177,7 @@ export function processQixPerfEvent(msg) {
         traverse_time: parseFloat(msg[20]),
         // Handle is either -1 or a number. Set to -99 if not a number
         handle: Number.isInteger(parseInt(msg[21], 10)) ? parseInt(msg[21], 10) : -99,
-        object_id: msg[22],
+        object_id: sanitizeField(msg[22], 100),
         // Positive integer, set to -1 if not am integer >= 0
         net_ram:
             Number.isInteger(parseInt(msg[23], 10)) && parseInt(msg[23], 10) >= 0
@@ -186,8 +187,8 @@ export function processQixPerfEvent(msg) {
             Number.isInteger(parseInt(msg[24], 10)) && parseInt(msg[24], 10) >= 0
                 ? parseInt(msg[24], 10)
                 : -1,
-        object_type: msg[25],
-        event_activity_source: eventActivitySource,
+        object_type: sanitizeField(msg[25], 100),
+        event_activity_source: sanitizeField(eventActivitySource, 50),
     };
 
     formatUserFields(msgObj);

src/lib/udp_handlers/log_events/handlers/repository-handler.js

@@ -4,6 +4,7 @@
 
 import globals from '../../../../globals.js';
 import { isoDateRegex, formatUserFields } from '../utils/common-utils.js';
+import { sanitizeField } from '../../../udp-queue-manager.js';
 
 /**
  * Process repository log events
@@ -33,22 +34,22 @@ export function processRepositoryEvent(msg) {
     globals.logger.verbose(`LOG EVENT: ${msg[0]}:${msg[5]}:${msg[4]}, ${msg[6]}, Msg: ${msg[8]}`);
 
     const msgObj = {
-        source: msg[0],
+        source: sanitizeField(msg[0], 100),
         log_row: Number.isInteger(parseInt(msg[1], 10)) ? parseInt(msg[1], 10) : -1,
-        ts_iso: isoDateRegex.test(msg[2]) ? msg[2] : '',
-        ts_local: isoDateRegex.test(msg[3]) ? msg[3] : '',
-        level: msg[4],
-        host: msg[5],
-        subsystem: msg[6],
-        windows_user: msg[7],
-        message: msg[8],
-        exception_message: msg[9],
-        user_directory: msg[10],
-        user_id: msg[11],
-        command: msg[12],
-        result_code: msg[13],
-        origin: msg[14],
-        context: msg[15],
+        ts_iso: isoDateRegex.test(msg[2]) ? sanitizeField(msg[2], 50) : '',
+        ts_local: isoDateRegex.test(msg[3]) ? sanitizeField(msg[3], 50) : '',
+        level: sanitizeField(msg[4], 20),
+        host: sanitizeField(msg[5], 100),
+        subsystem: sanitizeField(msg[6], 200),
+        windows_user: sanitizeField(msg[7], 100),
+        message: sanitizeField(msg[8], 1000),
+        exception_message: sanitizeField(msg[9], 1000),
+        user_directory: sanitizeField(msg[10], 100),
+        user_id: sanitizeField(msg[11], 100),
+        command: sanitizeField(msg[12], 200),
+        result_code: sanitizeField(msg[13], 50),
+        origin: sanitizeField(msg[14], 200),
+        context: sanitizeField(msg[15], 200),
     };
 
     formatUserFields(msgObj);

src/lib/udp_handlers/log_events/handlers/scheduler-handler.js

@@ -4,6 +4,7 @@
 
 import globals from '../../../../globals.js';
 import { isoDateRegex, uuidRegex, formatUserFields } from '../utils/common-utils.js';
+import { sanitizeField } from '../../../udp-queue-manager.js';
 
 /**
  * Process scheduler log events
@@ -35,21 +36,21 @@ export function processSchedulerEvent(msg) {
     globals.logger.verbose(`LOG EVENT: ${msg[0]}:${msg[5]}:${msg[4]}, ${msg[6]}, Msg: ${msg[8]}`);
 
     const msgObj = {
-        source: msg[0],
+        source: sanitizeField(msg[0], 100),
         log_row: Number.isInteger(parseInt(msg[1], 10)) ? parseInt(msg[1], 10) : -1,
-        ts_iso: isoDateRegex.test(msg[2]) ? msg[2] : '',
-        ts_local: isoDateRegex.test(msg[3]) ? msg[3] : '',
-        level: msg[4],
-        host: msg[5],
-        subsystem: msg[6],
-        windows_user: msg[7],
-        message: msg[8],
-        exception_message: msg[9],
-        user_directory: msg[10],
-        user_id: msg[11],
-        user_full: msg[12],
-        task_name: msg[13],
-        app_name: msg[14],
+        ts_iso: isoDateRegex.test(msg[2]) ? sanitizeField(msg[2], 50) : '',
+        ts_local: isoDateRegex.test(msg[3]) ? sanitizeField(msg[3], 50) : '',
+        level: sanitizeField(msg[4], 20),
+        host: sanitizeField(msg[5], 100),
+        subsystem: sanitizeField(msg[6], 200),
+        windows_user: sanitizeField(msg[7], 100),
+        message: sanitizeField(msg[8], 1000),
+        exception_message: sanitizeField(msg[9], 1000),
+        user_directory: sanitizeField(msg[10], 100),
+        user_id: sanitizeField(msg[11], 100),
+        user_full: sanitizeField(msg[12], 200),
+        task_name: sanitizeField(msg[13], 200),
+        app_name: sanitizeField(msg[14], 200),
         task_id: uuidRegex.test(msg[15]) ? msg[15] : '',
         app_id: uuidRegex.test(msg[16]) ? msg[16] : '',
         execution_id: uuidRegex.test(msg[17]) ? msg[17] : '',

src/lib/udp_handlers/user_events/__tests__/message-event.test.js

@@ -1,16 +1,19 @@
 // filepath: /Users/goran/code/butler-sos/src/lib/udp_handlers/user_events/__tests__/message-event.test.js
 import { jest, describe, test, expect, beforeEach, afterEach } from '@jest/globals';
 
-// Mock globals module
-jest.unstable_mockModule('../../../../globals.js', () => {
-    const mockGlobals = {
-        logger: {
-            error: jest.fn(),
-            warn: jest.fn(),
-            info: jest.fn(),
-            verbose: jest.fn(),
-            debug: jest.fn(),
-        },
+// Mock globals module - we only set up the structure, individual tests configure behavior
+const mockLogger = {
+    error: jest.fn(),
+    warn: jest.fn(),
+    info: jest.fn(),
+    verbose: jest.fn(),
+    debug: jest.fn(),
+    silly: jest.fn(),
+};
+
+jest.unstable_mockModule('../../../../globals.js', () => ({
+    default: {
+        logger: mockLogger,
         config: {
             get: jest.fn(),
             has: jest.fn(),
@@ -20,10 +23,8 @@ jest.unstable_mockModule('../../../../globals.js', () => {
         },
         appNames: [],
         getErrorMessage: jest.fn().mockImplementation((err) => err.toString()),
-    };
-
-    return { default: mockGlobals };
-});
+    },
+}));
 
 // Mock UAParser
 jest.unstable_mockModule('ua-parser-js', () => ({
@@ -112,7 +113,7 @@ describe('messageEventHandler', () => {
 
             await messageEventHandler(message, {});
 
-            expect(globals.logger.debug).toHaveBeenCalledWith(
+            expect(globals.logger.silly).toHaveBeenCalledWith(
                 expect.stringContaining('USER EVENT (raw):')
             );
             expect(globals.logger.verbose).toHaveBeenCalledWith(
@@ -133,7 +134,7 @@ describe('messageEventHandler', () => {
             await messageEventHandler(message, {});
 
             expect(globals.logger.verbose).toHaveBeenCalledWith(
-                'USER EVENT: /qseow-proxy-session/ - testuser2 - /app/87654321-4321-4321-4321-cba987654321'
+                expect.stringContaining('USER EVENT:')
             );
             expect(globals.udpEvents.addUserEvent).toHaveBeenCalledWith({
                 source: 'qseow-proxy-session',

src/lib/udp_handlers/user_events/message-event.js

@@ -3,6 +3,7 @@ import { UAParser } from 'ua-parser-js';
 
 // Load global variables and functions
 import globals from '../../../globals.js';
+import { sanitizeField } from '../../udp-queue-manager.js';
 import { postUserEventToInfluxdb } from '../../post-to-influxdb.js';
 import { postUserEventToNewRelic } from '../../post-to-new-relic.js';
 import { postUserEventToMQTT } from '../../post-to-mqtt.js';
@@ -120,14 +121,14 @@ export async function messageEventHandler(message, _remote) {
         let msgObj;
         if (msg[0] === 'qseow-proxy-connection' || msg[0] === 'qseow-proxy-session') {
             msgObj = {
-                messageType: msg[0],
-                host: msg[1],
-                command: msg[2],
-                user_directory: msg[3],
-                user_id: msg[4],
-                origin: msg[5],
-                context: msg[6],
-                message: msg[7],
+                messageType: sanitizeField(msg[0], 100),
+                host: sanitizeField(msg[1], 100),
+                command: sanitizeField(msg[2], 100),
+                user_directory: sanitizeField(msg[3], 100),
+                user_id: sanitizeField(msg[4], 100),
+                origin: sanitizeField(msg[5], 200),
+                context: sanitizeField(msg[6], 500),
+                message: sanitizeField(msg[7], 1000),
             };
 
             // Different log events deliver QSEoW user directory/user differently.