[Security Solution][Telemetry] Refactoring security telemetry task code (#114095)
Co-authored-by: Kibana Machine <42973632+kibanamachine@users.noreply.github.com>
This commit is contained in:
parent
a070aafe93
commit
fb4414786f
16 changed files with 866 additions and 1159 deletions
|
@ -0,0 +1,102 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License
|
||||
* 2.0; you may not use this file except in compliance with the Elastic License
|
||||
* 2.0.
|
||||
*/
|
||||
|
||||
import { ConcreteTaskInstance, TaskStatus } from '../../../../../task_manager/server';
|
||||
import { TelemetryEventsSender } from '../sender';
|
||||
import { TelemetryReceiver } from '../receiver';
|
||||
import { SecurityTelemetryTaskConfig } from '../task';
|
||||
import { PackagePolicy } from '../../../../../fleet/common/types/models/package_policy';
|
||||
|
||||
/**
|
||||
* Creates a mocked Telemetry Events Sender
|
||||
*/
|
||||
export const createMockTelemetryEventsSender = (
|
||||
enableTelemetry?: boolean
|
||||
): jest.Mocked<TelemetryEventsSender> => {
|
||||
return {
|
||||
setup: jest.fn(),
|
||||
start: jest.fn(),
|
||||
stop: jest.fn(),
|
||||
fetchTelemetryUrl: jest.fn(),
|
||||
queueTelemetryEvents: jest.fn(),
|
||||
processEvents: jest.fn(),
|
||||
isTelemetryOptedIn: jest.fn().mockReturnValue(enableTelemetry ?? jest.fn()),
|
||||
sendIfDue: jest.fn(),
|
||||
sendEvents: jest.fn(),
|
||||
} as unknown as jest.Mocked<TelemetryEventsSender>;
|
||||
};
|
||||
|
||||
export const createMockTelemetryReceiver = (
|
||||
diagnosticsAlert?: unknown
|
||||
): jest.Mocked<TelemetryReceiver> => {
|
||||
return {
|
||||
start: jest.fn(),
|
||||
fetchClusterInfo: jest.fn(),
|
||||
fetchLicenseInfo: jest.fn(),
|
||||
copyLicenseFields: jest.fn(),
|
||||
fetchFleetAgents: jest.fn(),
|
||||
fetchDiagnosticAlerts: jest.fn().mockReturnValue(diagnosticsAlert ?? jest.fn()),
|
||||
fetchEndpointMetrics: jest.fn(),
|
||||
fetchEndpointPolicyResponses: jest.fn(),
|
||||
fetchTrustedApplications: jest.fn(),
|
||||
fetchEndpointList: jest.fn(),
|
||||
fetchDetectionRules: jest.fn().mockReturnValue({ body: null }),
|
||||
} as unknown as jest.Mocked<TelemetryReceiver>;
|
||||
};
|
||||
|
||||
/**
|
||||
* Creates a mocked package policy
|
||||
*/
|
||||
export const createMockPackagePolicy = (): jest.Mocked<PackagePolicy> => {
|
||||
return {
|
||||
id: jest.fn(),
|
||||
inputs: jest.fn(),
|
||||
version: jest.fn(),
|
||||
revision: jest.fn(),
|
||||
updated_at: jest.fn(),
|
||||
updated_by: jest.fn(),
|
||||
created_at: jest.fn(),
|
||||
created_by: jest.fn(),
|
||||
} as unknown as jest.Mocked<PackagePolicy>;
|
||||
};
|
||||
|
||||
/**
|
||||
* Creates a mocked Security Telemetry Task Config
|
||||
*/
|
||||
export const createMockSecurityTelemetryTask = (
|
||||
testType?: string,
|
||||
testLastTimestamp?: string
|
||||
): jest.Mocked<SecurityTelemetryTaskConfig> => {
|
||||
return {
|
||||
type: testType,
|
||||
title: 'test title',
|
||||
interval: '0m',
|
||||
timeout: '0m',
|
||||
version: '0.0.0',
|
||||
getLastExecutionTime: jest.fn().mockReturnValue(testLastTimestamp ?? jest.fn()),
|
||||
runTask: jest.fn(),
|
||||
} as unknown as jest.Mocked<SecurityTelemetryTaskConfig>;
|
||||
};
|
||||
|
||||
/**
|
||||
* Creates a mocked Task Instance
|
||||
*/
|
||||
export const createMockTaskInstance = (testId: string, testType: string): ConcreteTaskInstance => {
|
||||
return {
|
||||
id: testId,
|
||||
runAt: new Date(),
|
||||
attempts: 0,
|
||||
ownerId: '',
|
||||
status: TaskStatus.Running,
|
||||
startedAt: new Date(),
|
||||
scheduledAt: new Date(),
|
||||
retryAt: new Date(),
|
||||
params: {},
|
||||
state: {},
|
||||
taskType: testType,
|
||||
} as ConcreteTaskInstance;
|
||||
};
|
|
@ -7,6 +7,12 @@
|
|||
|
||||
export const TELEMETRY_MAX_BUFFER_SIZE = 100;
|
||||
|
||||
export const MAX_SECURITY_LIST_TELEMETRY_BATCH = 100;
|
||||
|
||||
export const MAX_ENDPOINT_TELEMETRY_BATCH = 1_000;
|
||||
|
||||
export const MAX_DETECTION_RULE_TELEMETRY_BATCH = 1_000;
|
||||
|
||||
export const TELEMETRY_CHANNEL_LISTS = 'security-lists-v2';
|
||||
|
||||
export const TELEMETRY_CHANNEL_ENDPOINT_META = 'endpoint-metadata';
|
||||
|
|
|
@ -6,7 +6,7 @@
|
|||
*/
|
||||
|
||||
import moment from 'moment';
|
||||
import { createMockPackagePolicy } from './mocks';
|
||||
import { createMockPackagePolicy } from './__mocks__';
|
||||
import {
|
||||
LIST_DETECTION_RULE_EXCEPTION,
|
||||
LIST_ENDPOINT_EXCEPTION,
|
||||
|
|
|
@ -1,90 +0,0 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License
|
||||
* 2.0; you may not use this file except in compliance with the Elastic License
|
||||
* 2.0.
|
||||
*/
|
||||
|
||||
// eslint-disable-next-line max-classes-per-file
|
||||
import { TelemetryEventsSender } from './sender';
|
||||
import { TelemetryReceiver } from './receiver';
|
||||
import { DiagnosticTask, EndpointTask, ExceptionListsTask, DetectionRulesTask } from './tasks';
|
||||
import { PackagePolicy } from '../../../../fleet/common/types/models/package_policy';
|
||||
|
||||
/**
|
||||
* Creates a mocked Telemetry Events Sender
|
||||
*/
|
||||
export const createMockTelemetryEventsSender = (
|
||||
enableTelemtry: boolean
|
||||
): jest.Mocked<TelemetryEventsSender> => {
|
||||
return {
|
||||
setup: jest.fn(),
|
||||
start: jest.fn(),
|
||||
stop: jest.fn(),
|
||||
fetchTelemetryUrl: jest.fn(),
|
||||
queueTelemetryEvents: jest.fn(),
|
||||
processEvents: jest.fn(),
|
||||
isTelemetryOptedIn: jest.fn().mockReturnValue(enableTelemtry ?? jest.fn()),
|
||||
sendIfDue: jest.fn(),
|
||||
sendEvents: jest.fn(),
|
||||
} as unknown as jest.Mocked<TelemetryEventsSender>;
|
||||
};
|
||||
|
||||
export const createMockTelemetryReceiver = (): jest.Mocked<TelemetryReceiver> => {
|
||||
return {
|
||||
start: jest.fn(),
|
||||
fetchClusterInfo: jest.fn(),
|
||||
fetchLicenseInfo: jest.fn(),
|
||||
copyLicenseFields: jest.fn(),
|
||||
fetchDiagnosticAlerts: jest.fn(),
|
||||
fetchEndpointMetrics: jest.fn(),
|
||||
fetchEndpointPolicyResponses: jest.fn(),
|
||||
fetchTrustedApplications: jest.fn(),
|
||||
fetchDetectionRules: jest.fn(),
|
||||
fetchDetectionExceptionList: jest.fn(),
|
||||
} as unknown as jest.Mocked<TelemetryReceiver>;
|
||||
};
|
||||
|
||||
/**
|
||||
* Creates a mocked package policy
|
||||
*/
|
||||
export const createMockPackagePolicy = (): jest.Mocked<PackagePolicy> => {
|
||||
return {
|
||||
id: jest.fn(),
|
||||
inputs: jest.fn(),
|
||||
version: jest.fn(),
|
||||
revision: jest.fn(),
|
||||
updated_at: jest.fn(),
|
||||
updated_by: jest.fn(),
|
||||
created_at: jest.fn(),
|
||||
created_by: jest.fn(),
|
||||
} as unknown as jest.Mocked<PackagePolicy>;
|
||||
};
|
||||
|
||||
/**
|
||||
* Creates a mocked Telemetry Diagnostic Task
|
||||
*/
|
||||
export class MockTelemetryDiagnosticTask extends DiagnosticTask {
|
||||
public runTask = jest.fn();
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a mocked Telemetry Endpoint Task
|
||||
*/
|
||||
export class MockTelemetryEndpointTask extends EndpointTask {
|
||||
public runTask = jest.fn();
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a mocked Telemetry exception lists Task
|
||||
*/
|
||||
export class MockExceptionListsTask extends ExceptionListsTask {
|
||||
public runTask = jest.fn();
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a mocked Telemetry detection rules lists Task
|
||||
*/
|
||||
export class MockDetectionRuleListsTask extends DetectionRulesTask {
|
||||
public runTask = jest.fn();
|
||||
}
|
|
@ -18,10 +18,11 @@ import {
|
|||
} from '../../../../task_manager/server';
|
||||
import { TelemetryReceiver } from './receiver';
|
||||
import { allowlistEventFields, copyAllowlistedFields } from './filters';
|
||||
import { DiagnosticTask, EndpointTask, ExceptionListsTask, DetectionRulesTask } from './tasks';
|
||||
import { createTelemetryTaskConfigs } from './tasks';
|
||||
import { createUsageCounterLabel } from './helpers';
|
||||
import { TelemetryEvent } from './types';
|
||||
import { TELEMETRY_MAX_BUFFER_SIZE } from './constants';
|
||||
import { SecurityTelemetryTask, SecurityTelemetryTaskConfig } from './task';
|
||||
|
||||
const usageLabelPrefix: string[] = ['security_telemetry', 'sender'];
|
||||
|
||||
|
@ -39,10 +40,7 @@ export class TelemetryEventsSender {
|
|||
private isOptedIn?: boolean = true; // Assume true until the first check
|
||||
|
||||
private telemetryUsageCounter?: UsageCounter;
|
||||
private diagnosticTask?: DiagnosticTask;
|
||||
private endpointTask?: EndpointTask;
|
||||
private exceptionListsTask?: ExceptionListsTask;
|
||||
private detectionRulesTask?: DetectionRulesTask;
|
||||
private telemetryTasks?: SecurityTelemetryTask[];
|
||||
|
||||
constructor(logger: Logger) {
|
||||
this.logger = logger.get('telemetry_events');
|
||||
|
@ -58,19 +56,12 @@ export class TelemetryEventsSender {
|
|||
this.telemetryUsageCounter = telemetryUsageCounter;
|
||||
|
||||
if (taskManager) {
|
||||
this.diagnosticTask = new DiagnosticTask(this.logger, taskManager, this, telemetryReceiver);
|
||||
this.endpointTask = new EndpointTask(this.logger, taskManager, this, telemetryReceiver);
|
||||
this.detectionRulesTask = new DetectionRulesTask(
|
||||
this.logger,
|
||||
taskManager,
|
||||
this,
|
||||
telemetryReceiver
|
||||
);
|
||||
this.exceptionListsTask = new ExceptionListsTask(
|
||||
this.logger,
|
||||
taskManager,
|
||||
this,
|
||||
telemetryReceiver
|
||||
this.telemetryTasks = createTelemetryTaskConfigs().map(
|
||||
(config: SecurityTelemetryTaskConfig) => {
|
||||
const task = new SecurityTelemetryTask(config, this.logger, this, telemetryReceiver);
|
||||
task.register(taskManager);
|
||||
return task;
|
||||
}
|
||||
);
|
||||
}
|
||||
}
|
||||
|
@ -83,12 +74,9 @@ export class TelemetryEventsSender {
|
|||
this.telemetryStart = telemetryStart;
|
||||
this.receiver = receiver;
|
||||
|
||||
if (taskManager && this.diagnosticTask && this.endpointTask && this.exceptionListsTask) {
|
||||
this.logger.debug(`starting security telemetry tasks`);
|
||||
this.diagnosticTask.start(taskManager);
|
||||
this.endpointTask.start(taskManager);
|
||||
this.detectionRulesTask?.start(taskManager);
|
||||
this.exceptionListsTask?.start(taskManager);
|
||||
if (taskManager && this.telemetryTasks) {
|
||||
this.logger.debug(`Starting security telemetry tasks`);
|
||||
this.telemetryTasks.forEach((task) => task.start(taskManager));
|
||||
}
|
||||
|
||||
this.logger.debug(`Starting local task`);
|
||||
|
|
|
@ -0,0 +1,121 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License
|
||||
* 2.0; you may not use this file except in compliance with the Elastic License
|
||||
* 2.0.
|
||||
*/
|
||||
|
||||
import { loggingSystemMock } from 'src/core/server/mocks';
|
||||
import { taskManagerMock } from '../../../../task_manager/server/mocks';
|
||||
import { SuccessfulRunResult } from '../../../../task_manager/server/task';
|
||||
import { SecurityTelemetryTask } from './task';
|
||||
import {
|
||||
createMockTaskInstance,
|
||||
createMockTelemetryEventsSender,
|
||||
createMockTelemetryReceiver,
|
||||
createMockSecurityTelemetryTask,
|
||||
} from './__mocks__';
|
||||
|
||||
describe('test security telemetry task', () => {
|
||||
let logger: ReturnType<typeof loggingSystemMock.createLogger>;
|
||||
|
||||
beforeEach(() => {
|
||||
logger = loggingSystemMock.createLogger();
|
||||
});
|
||||
|
||||
test('telemetry task should be constructed', () => {
|
||||
const telemetryTask = new SecurityTelemetryTask(
|
||||
createMockSecurityTelemetryTask(),
|
||||
logger,
|
||||
createMockTelemetryEventsSender(true),
|
||||
createMockTelemetryReceiver()
|
||||
);
|
||||
|
||||
expect(telemetryTask).toBeInstanceOf(SecurityTelemetryTask);
|
||||
});
|
||||
|
||||
test('telemetry task should be registered and scheduled', async () => {
|
||||
const mockTaskManagerSetup = taskManagerMock.createSetup();
|
||||
const mockTaskManagerStart = taskManagerMock.createStart();
|
||||
const telemetryTask = new SecurityTelemetryTask(
|
||||
createMockSecurityTelemetryTask(),
|
||||
logger,
|
||||
createMockTelemetryEventsSender(true),
|
||||
createMockTelemetryReceiver()
|
||||
);
|
||||
telemetryTask.register(mockTaskManagerSetup);
|
||||
await telemetryTask.start(mockTaskManagerStart);
|
||||
|
||||
expect(mockTaskManagerSetup.registerTaskDefinitions).toHaveBeenCalled();
|
||||
expect(mockTaskManagerStart.ensureScheduled).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test('telemetry task should run if opted in', async () => {
|
||||
const {
|
||||
testLastTimestamp,
|
||||
testResult,
|
||||
telemetryTask,
|
||||
mockTelemetryTaskConfig,
|
||||
mockTelemetryEventsSender,
|
||||
mockTelemetryReceiver,
|
||||
} = await testTelemetryTaskRun(true);
|
||||
|
||||
expect(mockTelemetryTaskConfig.runTask).toHaveBeenCalledWith(
|
||||
telemetryTask.getTaskId(),
|
||||
logger,
|
||||
mockTelemetryReceiver,
|
||||
mockTelemetryEventsSender,
|
||||
{
|
||||
last: testLastTimestamp,
|
||||
current: testResult.state.lastExecutionTimestamp,
|
||||
}
|
||||
);
|
||||
});
|
||||
|
||||
test('telemetry task should not run if opted out', async () => {
|
||||
const { mockTelemetryTaskConfig } = await testTelemetryTaskRun(false);
|
||||
|
||||
expect(mockTelemetryTaskConfig.runTask).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
async function testTelemetryTaskRun(optedIn: boolean) {
|
||||
const now = new Date();
|
||||
const testType = 'security:test-task';
|
||||
const testLastTimestamp = now.toISOString();
|
||||
const mockTaskManagerSetup = taskManagerMock.createSetup();
|
||||
const mockTelemetryTaskConfig = createMockSecurityTelemetryTask(testType, testLastTimestamp);
|
||||
const mockTelemetryEventsSender = createMockTelemetryEventsSender(optedIn);
|
||||
const mockTelemetryReceiver = createMockTelemetryReceiver();
|
||||
const telemetryTask = new SecurityTelemetryTask(
|
||||
mockTelemetryTaskConfig,
|
||||
logger,
|
||||
mockTelemetryEventsSender,
|
||||
mockTelemetryReceiver
|
||||
);
|
||||
const mockTaskInstance = createMockTaskInstance(telemetryTask.getTaskId(), testType);
|
||||
|
||||
telemetryTask.register(mockTaskManagerSetup);
|
||||
expect(mockTaskManagerSetup.registerTaskDefinitions).toHaveBeenCalled();
|
||||
|
||||
const createTaskRunner =
|
||||
mockTaskManagerSetup.registerTaskDefinitions.mock.calls[0][0][testType].createTaskRunner;
|
||||
|
||||
const taskRunner = createTaskRunner({ taskInstance: mockTaskInstance });
|
||||
const testResult = (await taskRunner.run()) as SuccessfulRunResult;
|
||||
|
||||
expect(mockTelemetryTaskConfig.getLastExecutionTime).toHaveBeenCalled();
|
||||
expect(mockTelemetryEventsSender.isTelemetryOptedIn).toHaveBeenCalled();
|
||||
|
||||
expect(testResult).not.toBeNull();
|
||||
expect(testResult).toHaveProperty('state.lastExecutionTimestamp');
|
||||
|
||||
return {
|
||||
testLastTimestamp,
|
||||
testResult,
|
||||
telemetryTask,
|
||||
mockTelemetryTaskConfig,
|
||||
mockTelemetryEventsSender,
|
||||
mockTelemetryReceiver,
|
||||
};
|
||||
}
|
||||
});
|
148
x-pack/plugins/security_solution/server/lib/telemetry/task.ts
Normal file
148
x-pack/plugins/security_solution/server/lib/telemetry/task.ts
Normal file
|
@ -0,0 +1,148 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License
|
||||
* 2.0; you may not use this file except in compliance with the Elastic License
|
||||
* 2.0.
|
||||
*/
|
||||
|
||||
import moment from 'moment';
|
||||
import { Logger } from 'src/core/server';
|
||||
import {
|
||||
ConcreteTaskInstance,
|
||||
TaskManagerSetupContract,
|
||||
TaskManagerStartContract,
|
||||
} from '../../../../task_manager/server';
|
||||
import { TelemetryReceiver } from './receiver';
|
||||
import { TelemetryEventsSender } from './sender';
|
||||
|
||||
export interface SecurityTelemetryTaskConfig {
|
||||
type: string;
|
||||
title: string;
|
||||
interval: string;
|
||||
timeout: string;
|
||||
version: string;
|
||||
getLastExecutionTime?: LastExecutionTimestampCalculator;
|
||||
runTask: SecurityTelemetryTaskRunner;
|
||||
}
|
||||
|
||||
export type SecurityTelemetryTaskRunner = (
|
||||
taskId: string,
|
||||
logger: Logger,
|
||||
receiver: TelemetryReceiver,
|
||||
sender: TelemetryEventsSender,
|
||||
taskExecutionPeriod: TaskExecutionPeriod
|
||||
) => Promise<number>;
|
||||
|
||||
export interface TaskExecutionPeriod {
|
||||
last?: string;
|
||||
current: string;
|
||||
}
|
||||
|
||||
export type LastExecutionTimestampCalculator = (
|
||||
executeTo: string,
|
||||
lastExecutionTimestamp?: string
|
||||
) => string;
|
||||
|
||||
export class SecurityTelemetryTask {
|
||||
private readonly config: SecurityTelemetryTaskConfig;
|
||||
private readonly logger: Logger;
|
||||
private readonly sender: TelemetryEventsSender;
|
||||
private readonly receiver: TelemetryReceiver;
|
||||
|
||||
constructor(
|
||||
config: SecurityTelemetryTaskConfig,
|
||||
logger: Logger,
|
||||
sender: TelemetryEventsSender,
|
||||
receiver: TelemetryReceiver
|
||||
) {
|
||||
this.config = config;
|
||||
this.logger = logger;
|
||||
this.sender = sender;
|
||||
this.receiver = receiver;
|
||||
}
|
||||
|
||||
public getLastExecutionTime = (
|
||||
taskExecutionTime: string,
|
||||
taskInstance: ConcreteTaskInstance
|
||||
): string | undefined => {
|
||||
return this.config.getLastExecutionTime
|
||||
? this.config.getLastExecutionTime(
|
||||
taskExecutionTime,
|
||||
taskInstance.state?.lastExecutionTimestamp
|
||||
)
|
||||
: undefined;
|
||||
};
|
||||
|
||||
public getTaskId = (): string => {
|
||||
return `${this.config.type}:${this.config.version}`;
|
||||
};
|
||||
|
||||
public register = (taskManager: TaskManagerSetupContract) => {
|
||||
taskManager.registerTaskDefinitions({
|
||||
[this.config.type]: {
|
||||
title: this.config.title,
|
||||
timeout: this.config.timeout,
|
||||
createTaskRunner: ({ taskInstance }: { taskInstance: ConcreteTaskInstance }) => {
|
||||
const { state } = taskInstance;
|
||||
|
||||
return {
|
||||
run: async () => {
|
||||
const taskExecutionTime = moment().utc().toISOString();
|
||||
const executionPeriod = {
|
||||
last: this.getLastExecutionTime(taskExecutionTime, taskInstance),
|
||||
current: taskExecutionTime,
|
||||
};
|
||||
|
||||
const hits = await this.runTask(taskInstance.id, executionPeriod);
|
||||
|
||||
return {
|
||||
state: {
|
||||
lastExecutionTimestamp: taskExecutionTime,
|
||||
runs: (state.runs || 0) + 1,
|
||||
hits,
|
||||
},
|
||||
};
|
||||
},
|
||||
cancel: async () => {},
|
||||
};
|
||||
},
|
||||
},
|
||||
});
|
||||
};
|
||||
|
||||
public start = async (taskManager: TaskManagerStartContract) => {
|
||||
const taskId = this.getTaskId();
|
||||
this.logger.debug(`[task ${taskId}]: attempting to schedule`);
|
||||
try {
|
||||
await taskManager.ensureScheduled({
|
||||
id: taskId,
|
||||
taskType: this.config.type,
|
||||
scope: ['securitySolution'],
|
||||
schedule: {
|
||||
interval: this.config.interval,
|
||||
},
|
||||
state: { runs: 0 },
|
||||
params: { version: this.config.version },
|
||||
});
|
||||
} catch (e) {
|
||||
this.logger.error(`[task ${taskId}]: error scheduling task, received ${e.message}`);
|
||||
}
|
||||
};
|
||||
|
||||
public runTask = async (taskId: string, executionPeriod: TaskExecutionPeriod) => {
|
||||
this.logger.debug(`[task ${taskId}]: attempting to run`);
|
||||
if (taskId !== this.getTaskId()) {
|
||||
this.logger.debug(`[task ${taskId}]: outdated task`);
|
||||
return 0;
|
||||
}
|
||||
|
||||
const isOptedIn = await this.sender.isTelemetryOptedIn();
|
||||
if (!isOptedIn) {
|
||||
this.logger.debug(`[task ${taskId}]: telemetry is not opted-in`);
|
||||
return 0;
|
||||
}
|
||||
|
||||
this.logger.debug(`[task ${taskId}]: running task`);
|
||||
return this.config.runTask(taskId, this.logger, this.receiver, this.sender, executionPeriod);
|
||||
};
|
||||
}
|
|
@ -6,122 +6,33 @@
|
|||
*/
|
||||
|
||||
import { loggingSystemMock } from 'src/core/server/mocks';
|
||||
import { taskManagerMock } from '../../../../../task_manager/server/mocks';
|
||||
import { TaskStatus } from '../../../../../task_manager/server';
|
||||
import {
|
||||
TelemetryDetectionRulesTask,
|
||||
TelemetryDetectionRuleListsTaskConstants,
|
||||
} from './detection_rule';
|
||||
import {
|
||||
createMockTelemetryEventsSender,
|
||||
MockDetectionRuleListsTask,
|
||||
createMockTelemetryReceiver,
|
||||
} from '../mocks';
|
||||
import { createTelemetryDetectionRuleListsTaskConfig } from './detection_rule';
|
||||
import { createMockTelemetryEventsSender, createMockTelemetryReceiver } from '../__mocks__';
|
||||
|
||||
describe('test detection rule exception lists telemetry', () => {
|
||||
describe('security detection rule task test', () => {
|
||||
let logger: ReturnType<typeof loggingSystemMock.createLogger>;
|
||||
|
||||
beforeEach(() => {
|
||||
logger = loggingSystemMock.createLogger();
|
||||
});
|
||||
|
||||
describe('basic telemetry sanity checks', () => {
|
||||
test('detection rule lists task can register', () => {
|
||||
const telemetryDiagTask = new TelemetryDetectionRulesTask(
|
||||
logger,
|
||||
taskManagerMock.createSetup(),
|
||||
createMockTelemetryEventsSender(true),
|
||||
createMockTelemetryReceiver()
|
||||
);
|
||||
|
||||
expect(telemetryDiagTask).toBeInstanceOf(TelemetryDetectionRulesTask);
|
||||
});
|
||||
});
|
||||
|
||||
test('detection rule task should be registered', () => {
|
||||
const mockTaskManager = taskManagerMock.createSetup();
|
||||
new TelemetryDetectionRulesTask(
|
||||
logger,
|
||||
mockTaskManager,
|
||||
createMockTelemetryEventsSender(true),
|
||||
createMockTelemetryReceiver()
|
||||
);
|
||||
|
||||
expect(mockTaskManager.registerTaskDefinitions).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test('detection rule task should be scheduled', async () => {
|
||||
const mockTaskManagerSetup = taskManagerMock.createSetup();
|
||||
const telemetryDiagTask = new TelemetryDetectionRulesTask(
|
||||
logger,
|
||||
mockTaskManagerSetup,
|
||||
createMockTelemetryEventsSender(true),
|
||||
createMockTelemetryReceiver()
|
||||
);
|
||||
|
||||
const mockTaskManagerStart = taskManagerMock.createStart();
|
||||
await telemetryDiagTask.start(mockTaskManagerStart);
|
||||
expect(mockTaskManagerStart.ensureScheduled).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test('detection rule task should run', async () => {
|
||||
const mockContext = createMockTelemetryEventsSender(true);
|
||||
const mockTaskManager = taskManagerMock.createSetup();
|
||||
const mockReceiver = createMockTelemetryReceiver();
|
||||
const telemetryDiagTask = new MockDetectionRuleListsTask(
|
||||
logger,
|
||||
mockTaskManager,
|
||||
mockContext,
|
||||
mockReceiver
|
||||
);
|
||||
|
||||
const mockTaskInstance = {
|
||||
id: TelemetryDetectionRuleListsTaskConstants.TYPE,
|
||||
runAt: new Date(),
|
||||
attempts: 0,
|
||||
ownerId: '',
|
||||
status: TaskStatus.Running,
|
||||
startedAt: new Date(),
|
||||
scheduledAt: new Date(),
|
||||
retryAt: new Date(),
|
||||
params: {},
|
||||
state: {},
|
||||
taskType: TelemetryDetectionRuleListsTaskConstants.TYPE,
|
||||
test('security detecttion rule task should fetch security list data', async () => {
|
||||
const testTaskExecutionPeriod = {
|
||||
last: undefined,
|
||||
current: new Date().toISOString(),
|
||||
};
|
||||
const createTaskRunner =
|
||||
mockTaskManager.registerTaskDefinitions.mock.calls[0][0][
|
||||
TelemetryDetectionRuleListsTaskConstants.TYPE
|
||||
].createTaskRunner;
|
||||
const taskRunner = createTaskRunner({ taskInstance: mockTaskInstance });
|
||||
await taskRunner.run();
|
||||
expect(telemetryDiagTask.runTask).toHaveBeenCalled();
|
||||
});
|
||||
const mockTelemetryEventsSender = createMockTelemetryEventsSender();
|
||||
const mockTelemetryReceiver = createMockTelemetryReceiver();
|
||||
const telemetryDetectionRuleListsTaskConfig = createTelemetryDetectionRuleListsTaskConfig(1);
|
||||
|
||||
test('detection rule task should not query elastic if telemetry is not opted in', async () => {
|
||||
const mockSender = createMockTelemetryEventsSender(false);
|
||||
const mockTaskManager = taskManagerMock.createSetup();
|
||||
const mockReceiver = createMockTelemetryReceiver();
|
||||
new MockDetectionRuleListsTask(logger, mockTaskManager, mockSender, mockReceiver);
|
||||
await telemetryDetectionRuleListsTaskConfig.runTask(
|
||||
'test-id',
|
||||
logger,
|
||||
mockTelemetryReceiver,
|
||||
mockTelemetryEventsSender,
|
||||
testTaskExecutionPeriod
|
||||
);
|
||||
|
||||
const mockTaskInstance = {
|
||||
id: TelemetryDetectionRuleListsTaskConstants.TYPE,
|
||||
runAt: new Date(),
|
||||
attempts: 0,
|
||||
ownerId: '',
|
||||
status: TaskStatus.Running,
|
||||
startedAt: new Date(),
|
||||
scheduledAt: new Date(),
|
||||
retryAt: new Date(),
|
||||
params: {},
|
||||
state: {},
|
||||
taskType: TelemetryDetectionRuleListsTaskConstants.TYPE,
|
||||
};
|
||||
const createTaskRunner =
|
||||
mockTaskManager.registerTaskDefinitions.mock.calls[0][0][
|
||||
TelemetryDetectionRuleListsTaskConstants.TYPE
|
||||
].createTaskRunner;
|
||||
const taskRunner = createTaskRunner({ taskInstance: mockTaskInstance });
|
||||
await taskRunner.run();
|
||||
expect(mockReceiver.fetchDiagnosticAlerts).not.toHaveBeenCalled();
|
||||
expect(mockTelemetryReceiver.fetchDetectionRules).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
|
|
@ -5,145 +5,78 @@
|
|||
* 2.0.
|
||||
*/
|
||||
|
||||
import moment from 'moment';
|
||||
import { Logger } from 'src/core/server';
|
||||
import {
|
||||
ConcreteTaskInstance,
|
||||
TaskManagerSetupContract,
|
||||
TaskManagerStartContract,
|
||||
} from '../../../../../task_manager/server';
|
||||
import { LIST_DETECTION_RULE_EXCEPTION, TELEMETRY_CHANNEL_LISTS } from '../constants';
|
||||
import { batchTelemetryRecords, templateExceptionList } from '../helpers';
|
||||
import { TelemetryEventsSender } from '../sender';
|
||||
import { TelemetryReceiver } from '../receiver';
|
||||
import { ExceptionListItem, RuleSearchResult } from '../types';
|
||||
import { TaskExecutionPeriod } from '../task';
|
||||
|
||||
export const TelemetryDetectionRuleListsTaskConstants = {
|
||||
TIMEOUT: '10m',
|
||||
TYPE: 'security:telemetry-detection-rules',
|
||||
INTERVAL: '24h',
|
||||
VERSION: '1.0.0',
|
||||
};
|
||||
export function createTelemetryDetectionRuleListsTaskConfig(maxTelemetryBatch: number) {
|
||||
return {
|
||||
type: 'security:telemetry-detection-rules',
|
||||
title: 'Security Solution Detection Rule Lists Telemetry',
|
||||
interval: '24h',
|
||||
timeout: '10m',
|
||||
version: '1.0.0',
|
||||
runTask: async (
|
||||
taskId: string,
|
||||
logger: Logger,
|
||||
receiver: TelemetryReceiver,
|
||||
sender: TelemetryEventsSender,
|
||||
taskExecutionPeriod: TaskExecutionPeriod
|
||||
) => {
|
||||
// Lists Telemetry: Detection Rules
|
||||
|
||||
const MAX_TELEMETRY_BATCH = 1_000;
|
||||
const { body: prebuiltRules } = await receiver.fetchDetectionRules();
|
||||
|
||||
export class TelemetryDetectionRulesTask {
|
||||
private readonly logger: Logger;
|
||||
private readonly sender: TelemetryEventsSender;
|
||||
private readonly receiver: TelemetryReceiver;
|
||||
|
||||
constructor(
|
||||
logger: Logger,
|
||||
taskManager: TaskManagerSetupContract,
|
||||
sender: TelemetryEventsSender,
|
||||
receiver: TelemetryReceiver
|
||||
) {
|
||||
this.logger = logger;
|
||||
this.sender = sender;
|
||||
this.receiver = receiver;
|
||||
|
||||
taskManager.registerTaskDefinitions({
|
||||
[TelemetryDetectionRuleListsTaskConstants.TYPE]: {
|
||||
title: 'Security Solution Detection Rule Lists Telemetry',
|
||||
timeout: TelemetryDetectionRuleListsTaskConstants.TIMEOUT,
|
||||
createTaskRunner: ({ taskInstance }: { taskInstance: ConcreteTaskInstance }) => {
|
||||
const { state } = taskInstance;
|
||||
|
||||
return {
|
||||
run: async () => {
|
||||
const taskExecutionTime = moment().utc().toISOString();
|
||||
const hits = await this.runTask(taskInstance.id);
|
||||
|
||||
return {
|
||||
state: {
|
||||
lastExecutionTimestamp: taskExecutionTime,
|
||||
runs: (state.runs || 0) + 1,
|
||||
hits,
|
||||
},
|
||||
};
|
||||
},
|
||||
cancel: async () => {},
|
||||
};
|
||||
},
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
public start = async (taskManager: TaskManagerStartContract) => {
|
||||
try {
|
||||
await taskManager.ensureScheduled({
|
||||
id: this.getTaskId(),
|
||||
taskType: TelemetryDetectionRuleListsTaskConstants.TYPE,
|
||||
scope: ['securitySolution'],
|
||||
schedule: {
|
||||
interval: TelemetryDetectionRuleListsTaskConstants.INTERVAL,
|
||||
},
|
||||
state: { runs: 0 },
|
||||
params: { version: TelemetryDetectionRuleListsTaskConstants.VERSION },
|
||||
});
|
||||
} catch (e) {
|
||||
this.logger.error(`Error scheduling task, received ${e.message}`);
|
||||
}
|
||||
};
|
||||
|
||||
private getTaskId = (): string => {
|
||||
return `${TelemetryDetectionRuleListsTaskConstants.TYPE}:${TelemetryDetectionRuleListsTaskConstants.VERSION}`;
|
||||
};
|
||||
|
||||
public runTask = async (taskId: string) => {
|
||||
if (taskId !== this.getTaskId()) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
const isOptedIn = await this.sender.isTelemetryOptedIn();
|
||||
if (!isOptedIn) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
// Lists Telemetry: Detection Rules
|
||||
|
||||
const { body: prebuiltRules } = await this.receiver.fetchDetectionRules();
|
||||
|
||||
const cacheArray = prebuiltRules.hits.hits.reduce((cache, searchHit) => {
|
||||
const rule = searchHit._source as RuleSearchResult;
|
||||
const ruleId = rule.alert.params.ruleId;
|
||||
|
||||
const shouldNotProcess =
|
||||
rule === null ||
|
||||
rule === undefined ||
|
||||
ruleId === null ||
|
||||
ruleId === undefined ||
|
||||
searchHit._source?.alert.params.exceptionsList.length === 0;
|
||||
|
||||
if (shouldNotProcess) {
|
||||
return cache;
|
||||
if (!prebuiltRules) {
|
||||
logger.debug('no prebuilt rules found');
|
||||
return 0;
|
||||
}
|
||||
|
||||
cache.push(rule);
|
||||
return cache;
|
||||
}, [] as RuleSearchResult[]);
|
||||
const cacheArray = prebuiltRules.hits.hits.reduce((cache, searchHit) => {
|
||||
const rule = searchHit._source as RuleSearchResult;
|
||||
const ruleId = rule.alert.params.ruleId;
|
||||
|
||||
const detectionRuleExceptions = [] as ExceptionListItem[];
|
||||
for (const item of cacheArray) {
|
||||
const ruleVersion = item.alert.params.version;
|
||||
const shouldNotProcess =
|
||||
rule === null ||
|
||||
rule === undefined ||
|
||||
ruleId === null ||
|
||||
ruleId === undefined ||
|
||||
searchHit._source?.alert.params.exceptionsList.length === 0;
|
||||
|
||||
for (const ex of item.alert.params.exceptionsList) {
|
||||
const listItem = await this.receiver.fetchDetectionExceptionList(ex.list_id, ruleVersion);
|
||||
for (const exceptionItem of listItem.data) {
|
||||
detectionRuleExceptions.push(exceptionItem);
|
||||
if (shouldNotProcess) {
|
||||
return cache;
|
||||
}
|
||||
|
||||
cache.push(rule);
|
||||
return cache;
|
||||
}, [] as RuleSearchResult[]);
|
||||
|
||||
const detectionRuleExceptions = [] as ExceptionListItem[];
|
||||
for (const item of cacheArray) {
|
||||
const ruleVersion = item.alert.params.version;
|
||||
|
||||
for (const ex of item.alert.params.exceptionsList) {
|
||||
const listItem = await receiver.fetchDetectionExceptionList(ex.list_id, ruleVersion);
|
||||
for (const exceptionItem of listItem.data) {
|
||||
detectionRuleExceptions.push(exceptionItem);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const detectionRuleExceptionsJson = templateExceptionList(
|
||||
detectionRuleExceptions,
|
||||
LIST_DETECTION_RULE_EXCEPTION
|
||||
);
|
||||
const detectionRuleExceptionsJson = templateExceptionList(
|
||||
detectionRuleExceptions,
|
||||
LIST_DETECTION_RULE_EXCEPTION
|
||||
);
|
||||
|
||||
batchTelemetryRecords(detectionRuleExceptionsJson, MAX_TELEMETRY_BATCH).forEach((batch) => {
|
||||
this.sender.sendOnDemand(TELEMETRY_CHANNEL_LISTS, batch);
|
||||
});
|
||||
batchTelemetryRecords(detectionRuleExceptionsJson, maxTelemetryBatch).forEach((batch) => {
|
||||
sender.sendOnDemand(TELEMETRY_CHANNEL_LISTS, batch);
|
||||
});
|
||||
|
||||
return detectionRuleExceptions.length;
|
||||
return detectionRuleExceptions.length;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
|
|
@ -6,117 +6,45 @@
|
|||
*/
|
||||
|
||||
import { loggingSystemMock } from 'src/core/server/mocks';
|
||||
import { taskManagerMock } from '../../../../../task_manager/server/mocks';
|
||||
import { TaskStatus } from '../../../../../task_manager/server';
|
||||
import { TelemetryDiagTask, TelemetryDiagTaskConstants } from './diagnostic';
|
||||
import {
|
||||
createMockTelemetryEventsSender,
|
||||
MockTelemetryDiagnosticTask,
|
||||
createMockTelemetryReceiver,
|
||||
} from '../mocks';
|
||||
import { createTelemetryDiagnosticsTaskConfig } from './diagnostic';
|
||||
import { createMockTelemetryEventsSender, createMockTelemetryReceiver } from '../__mocks__';
|
||||
|
||||
describe('test', () => {
|
||||
describe('diagnostics telemetry task test', () => {
|
||||
let logger: ReturnType<typeof loggingSystemMock.createLogger>;
|
||||
|
||||
beforeEach(() => {
|
||||
logger = loggingSystemMock.createLogger();
|
||||
});
|
||||
|
||||
describe('basic diagnostic alert telemetry sanity checks', () => {
|
||||
test('diagnostic task can register', () => {
|
||||
const telemetryDiagTask = new TelemetryDiagTask(
|
||||
logger,
|
||||
taskManagerMock.createSetup(),
|
||||
createMockTelemetryEventsSender(true),
|
||||
createMockTelemetryReceiver()
|
||||
);
|
||||
|
||||
expect(telemetryDiagTask).toBeInstanceOf(TelemetryDiagTask);
|
||||
});
|
||||
});
|
||||
|
||||
test('diagnostic task should be registered', () => {
|
||||
const mockTaskManager = taskManagerMock.createSetup();
|
||||
new TelemetryDiagTask(
|
||||
logger,
|
||||
mockTaskManager,
|
||||
createMockTelemetryEventsSender(true),
|
||||
createMockTelemetryReceiver()
|
||||
);
|
||||
|
||||
expect(mockTaskManager.registerTaskDefinitions).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test('diagnostic task should be scheduled', async () => {
|
||||
const mockTaskManagerSetup = taskManagerMock.createSetup();
|
||||
const telemetryDiagTask = new TelemetryDiagTask(
|
||||
logger,
|
||||
mockTaskManagerSetup,
|
||||
createMockTelemetryEventsSender(true),
|
||||
createMockTelemetryReceiver()
|
||||
);
|
||||
|
||||
const mockTaskManagerStart = taskManagerMock.createStart();
|
||||
await telemetryDiagTask.start(mockTaskManagerStart);
|
||||
expect(mockTaskManagerStart.ensureScheduled).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test('diagnostic task should run', async () => {
|
||||
const mockContext = createMockTelemetryEventsSender(true);
|
||||
const mockTaskManager = taskManagerMock.createSetup();
|
||||
const mockReceiver = createMockTelemetryReceiver();
|
||||
const telemetryDiagTask = new MockTelemetryDiagnosticTask(
|
||||
logger,
|
||||
mockTaskManager,
|
||||
mockContext,
|
||||
mockReceiver
|
||||
);
|
||||
|
||||
const mockTaskInstance = {
|
||||
id: TelemetryDiagTaskConstants.TYPE,
|
||||
runAt: new Date(),
|
||||
attempts: 0,
|
||||
ownerId: '',
|
||||
status: TaskStatus.Running,
|
||||
startedAt: new Date(),
|
||||
scheduledAt: new Date(),
|
||||
retryAt: new Date(),
|
||||
params: {},
|
||||
state: {},
|
||||
taskType: TelemetryDiagTaskConstants.TYPE,
|
||||
test('diagnostics telemetry task should query and enqueue events', async () => {
|
||||
const testAlertDoc1 = { id: 'test1' };
|
||||
const testAlertDoc2 = { id: 'test2' };
|
||||
const testDiagnosticsAlerts = {
|
||||
hits: { hits: [{ _source: [testAlertDoc1] }, { _source: [testAlertDoc2] }] },
|
||||
};
|
||||
const createTaskRunner =
|
||||
mockTaskManager.registerTaskDefinitions.mock.calls[0][0][TelemetryDiagTaskConstants.TYPE]
|
||||
.createTaskRunner;
|
||||
const taskRunner = createTaskRunner({ taskInstance: mockTaskInstance });
|
||||
await taskRunner.run();
|
||||
expect(telemetryDiagTask.runTask).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test('diagnostic task should not query elastic if telemetry is not opted in', async () => {
|
||||
const mockSender = createMockTelemetryEventsSender(false);
|
||||
const mockTaskManager = taskManagerMock.createSetup();
|
||||
const mockReceiver = createMockTelemetryReceiver();
|
||||
new MockTelemetryDiagnosticTask(logger, mockTaskManager, mockSender, mockReceiver);
|
||||
|
||||
const mockTaskInstance = {
|
||||
id: TelemetryDiagTaskConstants.TYPE,
|
||||
runAt: new Date(),
|
||||
attempts: 0,
|
||||
ownerId: '',
|
||||
status: TaskStatus.Running,
|
||||
startedAt: new Date(),
|
||||
scheduledAt: new Date(),
|
||||
retryAt: new Date(),
|
||||
params: {},
|
||||
state: {},
|
||||
taskType: TelemetryDiagTaskConstants.TYPE,
|
||||
const testTaskExecutionPeriod = {
|
||||
last: new Date().toISOString(),
|
||||
current: new Date().toISOString(),
|
||||
};
|
||||
const createTaskRunner =
|
||||
mockTaskManager.registerTaskDefinitions.mock.calls[0][0][TelemetryDiagTaskConstants.TYPE]
|
||||
.createTaskRunner;
|
||||
const taskRunner = createTaskRunner({ taskInstance: mockTaskInstance });
|
||||
await taskRunner.run();
|
||||
expect(mockReceiver.fetchDiagnosticAlerts).not.toHaveBeenCalled();
|
||||
const mockTelemetryEventsSender = createMockTelemetryEventsSender();
|
||||
const mockTelemetryReceiver = createMockTelemetryReceiver(testDiagnosticsAlerts);
|
||||
const telemetryDiagnoticsTaskConfig = createTelemetryDiagnosticsTaskConfig();
|
||||
|
||||
await telemetryDiagnoticsTaskConfig.runTask(
|
||||
'test-id',
|
||||
logger,
|
||||
mockTelemetryReceiver,
|
||||
mockTelemetryEventsSender,
|
||||
testTaskExecutionPeriod
|
||||
);
|
||||
|
||||
expect(mockTelemetryReceiver.fetchDiagnosticAlerts).toHaveBeenCalledWith(
|
||||
testTaskExecutionPeriod.last,
|
||||
testTaskExecutionPeriod.current
|
||||
);
|
||||
|
||||
expect(mockTelemetryEventsSender.queueTelemetryEvents).toHaveBeenCalledWith(
|
||||
testDiagnosticsAlerts.hits.hits.flatMap((doc) => [doc._source])
|
||||
);
|
||||
});
|
||||
});
|
||||
|
|
|
@ -5,118 +5,49 @@
|
|||
* 2.0.
|
||||
*/
|
||||
|
||||
import moment from 'moment';
|
||||
import { Logger } from 'src/core/server';
|
||||
import {
|
||||
ConcreteTaskInstance,
|
||||
TaskManagerSetupContract,
|
||||
TaskManagerStartContract,
|
||||
} from '../../../../../task_manager/server';
|
||||
import { getPreviousDiagTaskTimestamp } from '../helpers';
|
||||
import { TelemetryEventsSender } from '../sender';
|
||||
import { TelemetryEvent } from '../types';
|
||||
import { TelemetryReceiver } from '../receiver';
|
||||
import { TaskExecutionPeriod } from '../task';
|
||||
|
||||
export const TelemetryDiagTaskConstants = {
|
||||
TIMEOUT: '1m',
|
||||
TYPE: 'security:endpoint-diagnostics',
|
||||
INTERVAL: '5m',
|
||||
VERSION: '1.0.0',
|
||||
};
|
||||
export function createTelemetryDiagnosticsTaskConfig() {
|
||||
return {
|
||||
type: 'security:endpoint-diagnostics',
|
||||
title: 'Security Solution Telemetry Diagnostics task',
|
||||
interval: '5m',
|
||||
timeout: '1m',
|
||||
version: '1.0.0',
|
||||
getLastExecutionTime: getPreviousDiagTaskTimestamp,
|
||||
runTask: async (
|
||||
taskId: string,
|
||||
logger: Logger,
|
||||
receiver: TelemetryReceiver,
|
||||
sender: TelemetryEventsSender,
|
||||
taskExecutionPeriod: TaskExecutionPeriod
|
||||
) => {
|
||||
if (!taskExecutionPeriod.last) {
|
||||
throw new Error('last execution timestamp is required');
|
||||
}
|
||||
|
||||
export class TelemetryDiagTask {
|
||||
private readonly logger: Logger;
|
||||
private readonly sender: TelemetryEventsSender;
|
||||
private readonly receiver: TelemetryReceiver;
|
||||
const response = await receiver.fetchDiagnosticAlerts(
|
||||
taskExecutionPeriod.last,
|
||||
taskExecutionPeriod.current
|
||||
);
|
||||
|
||||
constructor(
|
||||
logger: Logger,
|
||||
taskManager: TaskManagerSetupContract,
|
||||
sender: TelemetryEventsSender,
|
||||
receiver: TelemetryReceiver
|
||||
) {
|
||||
this.logger = logger;
|
||||
this.sender = sender;
|
||||
this.receiver = receiver;
|
||||
const hits = response.hits?.hits || [];
|
||||
if (!Array.isArray(hits) || !hits.length) {
|
||||
logger.debug('no diagnostic alerts retrieved');
|
||||
return 0;
|
||||
}
|
||||
logger.debug(`Received ${hits.length} diagnostic alerts`);
|
||||
|
||||
taskManager.registerTaskDefinitions({
|
||||
[TelemetryDiagTaskConstants.TYPE]: {
|
||||
title: 'Security Solution Telemetry Diagnostics task',
|
||||
timeout: TelemetryDiagTaskConstants.TIMEOUT,
|
||||
createTaskRunner: ({ taskInstance }: { taskInstance: ConcreteTaskInstance }) => {
|
||||
const { state } = taskInstance;
|
||||
|
||||
return {
|
||||
run: async () => {
|
||||
const executeTo = moment().utc().toISOString();
|
||||
const executeFrom = getPreviousDiagTaskTimestamp(
|
||||
executeTo,
|
||||
taskInstance.state?.lastExecutionTimestamp
|
||||
);
|
||||
const hits = await this.runTask(taskInstance.id, executeFrom, executeTo);
|
||||
|
||||
return {
|
||||
state: {
|
||||
lastExecutionTimestamp: executeTo,
|
||||
lastDiagAlertCount: hits,
|
||||
runs: (state.runs || 0) + 1,
|
||||
},
|
||||
};
|
||||
},
|
||||
cancel: async () => {},
|
||||
};
|
||||
},
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
public start = async (taskManager: TaskManagerStartContract) => {
|
||||
try {
|
||||
await taskManager.ensureScheduled({
|
||||
id: this.getTaskId(),
|
||||
taskType: TelemetryDiagTaskConstants.TYPE,
|
||||
scope: ['securitySolution'],
|
||||
schedule: {
|
||||
interval: TelemetryDiagTaskConstants.INTERVAL,
|
||||
},
|
||||
state: { runs: 0 },
|
||||
params: { version: TelemetryDiagTaskConstants.VERSION },
|
||||
});
|
||||
} catch (e) {
|
||||
this.logger.error(`Error scheduling task, received ${e.message}`);
|
||||
}
|
||||
};
|
||||
|
||||
private getTaskId = (): string => {
|
||||
return `${TelemetryDiagTaskConstants.TYPE}:${TelemetryDiagTaskConstants.VERSION}`;
|
||||
};
|
||||
|
||||
public runTask = async (taskId: string, searchFrom: string, searchTo: string) => {
|
||||
this.logger.debug(`Running task ${taskId}`);
|
||||
if (taskId !== this.getTaskId()) {
|
||||
this.logger.debug(`Outdated task running: ${taskId}`);
|
||||
return 0;
|
||||
}
|
||||
|
||||
const isOptedIn = await this.sender.isTelemetryOptedIn();
|
||||
if (!isOptedIn) {
|
||||
this.logger.debug(`Telemetry is not opted-in.`);
|
||||
return 0;
|
||||
}
|
||||
|
||||
const response = await this.receiver.fetchDiagnosticAlerts(searchFrom, searchTo);
|
||||
|
||||
const hits = response.hits?.hits || [];
|
||||
if (!Array.isArray(hits) || !hits.length) {
|
||||
this.logger.debug('no diagnostic alerts retrieved');
|
||||
return 0;
|
||||
}
|
||||
this.logger.debug(`Received ${hits.length} diagnostic alerts`);
|
||||
|
||||
const diagAlerts: TelemetryEvent[] = hits.flatMap((h) =>
|
||||
h._source != null ? [h._source] : []
|
||||
);
|
||||
this.sender.queueTelemetryEvents(diagAlerts);
|
||||
return diagAlerts.length;
|
||||
const diagAlerts: TelemetryEvent[] = hits.flatMap((h) =>
|
||||
h._source != null ? [h._source] : []
|
||||
);
|
||||
sender.queueTelemetryEvents(diagAlerts);
|
||||
return diagAlerts.length;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
|
|
@ -6,119 +6,41 @@
|
|||
*/
|
||||
|
||||
import { loggingSystemMock } from 'src/core/server/mocks';
|
||||
import { TaskStatus } from '../../../../../task_manager/server';
|
||||
import { taskManagerMock } from '../../../../../task_manager/server/mocks';
|
||||
import { createTelemetryEndpointTaskConfig } from './endpoint';
|
||||
import { createMockTelemetryEventsSender, createMockTelemetryReceiver } from '../__mocks__';
|
||||
|
||||
import { TelemetryEndpointTask, TelemetryEndpointTaskConstants } from './endpoint';
|
||||
import {
|
||||
createMockTelemetryEventsSender,
|
||||
MockTelemetryEndpointTask,
|
||||
createMockTelemetryReceiver,
|
||||
} from '../mocks';
|
||||
|
||||
describe('test', () => {
|
||||
describe('endpoint telemetry task test', () => {
|
||||
let logger: ReturnType<typeof loggingSystemMock.createLogger>;
|
||||
|
||||
beforeEach(() => {
|
||||
logger = loggingSystemMock.createLogger();
|
||||
});
|
||||
|
||||
describe('endpoint alert telemetry checks', () => {
|
||||
test('the endpoint task can register', () => {
|
||||
const telemetryEndpointTask = new TelemetryEndpointTask(
|
||||
logger,
|
||||
taskManagerMock.createSetup(),
|
||||
createMockTelemetryEventsSender(true),
|
||||
createMockTelemetryReceiver()
|
||||
);
|
||||
|
||||
expect(telemetryEndpointTask).toBeInstanceOf(TelemetryEndpointTask);
|
||||
});
|
||||
});
|
||||
|
||||
test('the endpoint task should be registered', () => {
|
||||
const mockTaskManager = taskManagerMock.createSetup();
|
||||
new TelemetryEndpointTask(
|
||||
logger,
|
||||
mockTaskManager,
|
||||
createMockTelemetryEventsSender(true),
|
||||
createMockTelemetryReceiver()
|
||||
);
|
||||
|
||||
expect(mockTaskManager.registerTaskDefinitions).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test('the endpoint task should be scheduled', async () => {
|
||||
const mockTaskManagerSetup = taskManagerMock.createSetup();
|
||||
const telemetryEndpointTask = new TelemetryEndpointTask(
|
||||
logger,
|
||||
mockTaskManagerSetup,
|
||||
createMockTelemetryEventsSender(true),
|
||||
createMockTelemetryReceiver()
|
||||
);
|
||||
|
||||
const mockTaskManagerStart = taskManagerMock.createStart();
|
||||
await telemetryEndpointTask.start(mockTaskManagerStart);
|
||||
expect(mockTaskManagerStart.ensureScheduled).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test('endpoint task should not query elastic if telemetry is not opted in', async () => {
|
||||
const mockSender = createMockTelemetryEventsSender(false);
|
||||
const mockTaskManager = taskManagerMock.createSetup();
|
||||
const mockReceiver = createMockTelemetryReceiver();
|
||||
new MockTelemetryEndpointTask(logger, mockTaskManager, mockSender, mockReceiver);
|
||||
|
||||
const mockTaskInstance = {
|
||||
id: TelemetryEndpointTaskConstants.TYPE,
|
||||
runAt: new Date(),
|
||||
attempts: 0,
|
||||
ownerId: '',
|
||||
status: TaskStatus.Running,
|
||||
startedAt: new Date(),
|
||||
scheduledAt: new Date(),
|
||||
retryAt: new Date(),
|
||||
params: {},
|
||||
state: {},
|
||||
taskType: TelemetryEndpointTaskConstants.TYPE,
|
||||
test('endpoint telemetry task should fetch endpoint data', async () => {
|
||||
const testTaskExecutionPeriod = {
|
||||
last: new Date().toISOString(),
|
||||
current: new Date().toISOString(),
|
||||
};
|
||||
const createTaskRunner =
|
||||
mockTaskManager.registerTaskDefinitions.mock.calls[0][0][TelemetryEndpointTaskConstants.TYPE]
|
||||
.createTaskRunner;
|
||||
const taskRunner = createTaskRunner({ taskInstance: mockTaskInstance });
|
||||
await taskRunner.run();
|
||||
expect(mockReceiver.fetchEndpointMetrics).not.toHaveBeenCalled();
|
||||
expect(mockReceiver.fetchEndpointPolicyResponses).not.toHaveBeenCalled();
|
||||
});
|
||||
const mockTelemetryEventsSender = createMockTelemetryEventsSender();
|
||||
const mockTelemetryReceiver = createMockTelemetryReceiver();
|
||||
const telemetryEndpointTaskConfig = createTelemetryEndpointTaskConfig(1);
|
||||
|
||||
test('endpoint task should run when opted in', async () => {
|
||||
const mockSender = createMockTelemetryEventsSender(true);
|
||||
const mockTaskManager = taskManagerMock.createSetup();
|
||||
const mockReceiver = createMockTelemetryReceiver();
|
||||
const telemetryEpMetaTask = new MockTelemetryEndpointTask(
|
||||
await telemetryEndpointTaskConfig.runTask(
|
||||
'test-id',
|
||||
logger,
|
||||
mockTaskManager,
|
||||
mockSender,
|
||||
mockReceiver
|
||||
mockTelemetryReceiver,
|
||||
mockTelemetryEventsSender,
|
||||
testTaskExecutionPeriod
|
||||
);
|
||||
|
||||
const mockTaskInstance = {
|
||||
id: TelemetryEndpointTaskConstants.TYPE,
|
||||
runAt: new Date(),
|
||||
attempts: 0,
|
||||
ownerId: '',
|
||||
status: TaskStatus.Running,
|
||||
startedAt: new Date(),
|
||||
scheduledAt: new Date(),
|
||||
retryAt: new Date(),
|
||||
params: {},
|
||||
state: {},
|
||||
taskType: TelemetryEndpointTaskConstants.TYPE,
|
||||
};
|
||||
const createTaskRunner =
|
||||
mockTaskManager.registerTaskDefinitions.mock.calls[0][0][TelemetryEndpointTaskConstants.TYPE]
|
||||
.createTaskRunner;
|
||||
const taskRunner = createTaskRunner({ taskInstance: mockTaskInstance });
|
||||
await taskRunner.run();
|
||||
expect(telemetryEpMetaTask.runTask).toHaveBeenCalled();
|
||||
expect(mockTelemetryReceiver.fetchFleetAgents).toHaveBeenCalled();
|
||||
expect(mockTelemetryReceiver.fetchEndpointMetrics).toHaveBeenCalledWith(
|
||||
testTaskExecutionPeriod.last,
|
||||
testTaskExecutionPeriod.current
|
||||
);
|
||||
expect(mockTelemetryReceiver.fetchEndpointPolicyResponses).toHaveBeenCalledWith(
|
||||
testTaskExecutionPeriod.last,
|
||||
testTaskExecutionPeriod.current
|
||||
);
|
||||
});
|
||||
});
|
||||
|
|
|
@ -5,35 +5,23 @@
|
|||
* 2.0.
|
||||
*/
|
||||
|
||||
import moment from 'moment';
|
||||
import { Logger } from 'src/core/server';
|
||||
import {
|
||||
ConcreteTaskInstance,
|
||||
TaskManagerSetupContract,
|
||||
TaskManagerStartContract,
|
||||
} from '../../../../../task_manager/server';
|
||||
import {
|
||||
batchTelemetryRecords,
|
||||
getPreviousDailyTaskTimestamp,
|
||||
isPackagePolicyList,
|
||||
} from '../helpers';
|
||||
import { TelemetryEventsSender } from '../sender';
|
||||
import { PolicyData } from '../../../../common/endpoint/types';
|
||||
import { FLEET_ENDPOINT_PACKAGE } from '../../../../../fleet/common';
|
||||
import {
|
||||
EndpointMetricsAggregation,
|
||||
EndpointPolicyResponseAggregation,
|
||||
EndpointPolicyResponseDocument,
|
||||
} from '../types';
|
||||
import { TELEMETRY_CHANNEL_ENDPOINT_META } from '../constants';
|
||||
import { TelemetryReceiver } from '../receiver';
|
||||
|
||||
export const TelemetryEndpointTaskConstants = {
|
||||
TIMEOUT: '5m',
|
||||
TYPE: 'security:endpoint-meta-telemetry',
|
||||
INTERVAL: '24h',
|
||||
VERSION: '1.0.0',
|
||||
};
|
||||
import { TaskExecutionPeriod } from '../task';
|
||||
import {
|
||||
batchTelemetryRecords,
|
||||
getPreviousDailyTaskTimestamp,
|
||||
isPackagePolicyList,
|
||||
} from '../helpers';
|
||||
import { PolicyData } from '../../../../common/endpoint/types';
|
||||
import { FLEET_ENDPOINT_PACKAGE } from '../../../../../fleet/common';
|
||||
import { TELEMETRY_CHANNEL_ENDPOINT_META } from '../constants';
|
||||
|
||||
// Endpoint agent uses this Policy ID while it's installing.
|
||||
const DefaultEndpointPolicyIdToIgnore = '00000000-0000-0000-0000-000000000000';
|
||||
|
@ -45,295 +33,236 @@ const EmptyFleetAgentResponse = {
|
|||
perPage: 0,
|
||||
};
|
||||
|
||||
/** Telemetry Endpoint Task
|
||||
*
|
||||
* The Endpoint Telemetry task is a daily batch job that collects and transmits non-sensitive
|
||||
* endpoint performance and policy logs to Elastic Security Data Engineering. It is used to
|
||||
* identify bugs or common UX issues with the Elastic Security Endpoint agent.
|
||||
*/
|
||||
export class TelemetryEndpointTask {
|
||||
private readonly logger: Logger;
|
||||
private readonly sender: TelemetryEventsSender;
|
||||
private readonly receiver: TelemetryReceiver;
|
||||
export function createTelemetryEndpointTaskConfig(maxTelemetryBatch: number) {
|
||||
return {
|
||||
type: 'security:endpoint-meta-telemetry',
|
||||
title: 'Security Solution Telemetry Endpoint Metrics and Info task',
|
||||
interval: '24h',
|
||||
timeout: '5m',
|
||||
version: '1.0.0',
|
||||
getLastExecutionTime: getPreviousDailyTaskTimestamp,
|
||||
runTask: async (
|
||||
taskId: string,
|
||||
logger: Logger,
|
||||
receiver: TelemetryReceiver,
|
||||
sender: TelemetryEventsSender,
|
||||
taskExecutionPeriod: TaskExecutionPeriod
|
||||
) => {
|
||||
if (!taskExecutionPeriod.last) {
|
||||
throw new Error('last execution timestamp is required');
|
||||
}
|
||||
|
||||
constructor(
|
||||
logger: Logger,
|
||||
taskManager: TaskManagerSetupContract,
|
||||
sender: TelemetryEventsSender,
|
||||
receiver: TelemetryReceiver
|
||||
) {
|
||||
this.logger = logger;
|
||||
this.sender = sender;
|
||||
this.receiver = receiver;
|
||||
const endpointData = await fetchEndpointData(
|
||||
receiver,
|
||||
taskExecutionPeriod.last,
|
||||
taskExecutionPeriod.current
|
||||
);
|
||||
|
||||
taskManager.registerTaskDefinitions({
|
||||
[TelemetryEndpointTaskConstants.TYPE]: {
|
||||
title: 'Security Solution Telemetry Endpoint Metrics and Info task',
|
||||
timeout: TelemetryEndpointTaskConstants.TIMEOUT,
|
||||
createTaskRunner: ({ taskInstance }: { taskInstance: ConcreteTaskInstance }) => {
|
||||
const { state } = taskInstance;
|
||||
/** STAGE 1 - Fetch Endpoint Agent Metrics
|
||||
*
|
||||
* Reads Endpoint Agent metrics out of the `.ds-metrics-endpoint.metrics` data stream
|
||||
* and buckets them by Endpoint Agent id and sorts by the top hit. The EP agent will
|
||||
* report its metrics once per day OR every time a policy change has occured. If
|
||||
* a metric document(s) exists for an EP agent we map to fleet agent and policy
|
||||
*/
|
||||
if (endpointData.endpointMetrics === undefined) {
|
||||
logger.debug(`no endpoint metrics to report`);
|
||||
return 0;
|
||||
}
|
||||
|
||||
const { body: endpointMetricsResponse } = endpointData.endpointMetrics as unknown as {
|
||||
body: EndpointMetricsAggregation;
|
||||
};
|
||||
|
||||
if (endpointMetricsResponse.aggregations === undefined) {
|
||||
logger.debug(`no endpoint metrics to report`);
|
||||
return 0;
|
||||
}
|
||||
|
||||
const endpointMetrics = endpointMetricsResponse.aggregations.endpoint_agents.buckets.map(
|
||||
(epMetrics) => {
|
||||
return {
|
||||
run: async () => {
|
||||
const taskExecutionTime = moment().utc().toISOString();
|
||||
const lastExecutionTimestamp = getPreviousDailyTaskTimestamp(
|
||||
taskExecutionTime,
|
||||
taskInstance.state?.lastExecutionTimestamp
|
||||
);
|
||||
|
||||
const hits = await this.runTask(
|
||||
taskInstance.id,
|
||||
lastExecutionTimestamp,
|
||||
taskExecutionTime
|
||||
);
|
||||
|
||||
return {
|
||||
state: {
|
||||
lastExecutionTimestamp: taskExecutionTime,
|
||||
runs: (state.runs || 0) + 1,
|
||||
hits,
|
||||
},
|
||||
};
|
||||
},
|
||||
cancel: async () => {},
|
||||
endpoint_agent: epMetrics.latest_metrics.hits.hits[0]._source.agent.id,
|
||||
endpoint_version: epMetrics.latest_metrics.hits.hits[0]._source.agent.version,
|
||||
endpoint_metrics: epMetrics.latest_metrics.hits.hits[0]._source,
|
||||
};
|
||||
},
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
public start = async (taskManager: TaskManagerStartContract) => {
|
||||
try {
|
||||
await taskManager.ensureScheduled({
|
||||
id: this.getTaskId(),
|
||||
taskType: TelemetryEndpointTaskConstants.TYPE,
|
||||
scope: ['securitySolution'],
|
||||
schedule: {
|
||||
interval: TelemetryEndpointTaskConstants.INTERVAL,
|
||||
},
|
||||
state: { runs: 0 },
|
||||
params: { version: TelemetryEndpointTaskConstants.VERSION },
|
||||
});
|
||||
} catch (e) {
|
||||
this.logger.error(`Error scheduling task, received ${e.message}`);
|
||||
}
|
||||
};
|
||||
|
||||
private getTaskId = (): string => {
|
||||
return `${TelemetryEndpointTaskConstants.TYPE}:${TelemetryEndpointTaskConstants.VERSION}`;
|
||||
};
|
||||
|
||||
private async fetchEndpointData(executeFrom: string, executeTo: string) {
|
||||
const [fleetAgentsResponse, epMetricsResponse, policyResponse] = await Promise.allSettled([
|
||||
this.receiver.fetchFleetAgents(),
|
||||
this.receiver.fetchEndpointMetrics(executeFrom, executeTo),
|
||||
this.receiver.fetchEndpointPolicyResponses(executeFrom, executeTo),
|
||||
]);
|
||||
|
||||
return {
|
||||
fleetAgentsResponse:
|
||||
fleetAgentsResponse.status === 'fulfilled'
|
||||
? fleetAgentsResponse.value
|
||||
: EmptyFleetAgentResponse,
|
||||
endpointMetrics:
|
||||
epMetricsResponse.status === 'fulfilled' ? epMetricsResponse.value : undefined,
|
||||
epPolicyResponse: policyResponse.status === 'fulfilled' ? policyResponse.value : undefined,
|
||||
};
|
||||
}
|
||||
|
||||
public runTask = async (taskId: string, executeFrom: string, executeTo: string) => {
|
||||
if (taskId !== this.getTaskId()) {
|
||||
this.logger.debug(`Outdated task running: ${taskId}`);
|
||||
return 0;
|
||||
}
|
||||
|
||||
const isOptedIn = await this.sender.isTelemetryOptedIn();
|
||||
if (!isOptedIn) {
|
||||
this.logger.debug(`Telemetry is not opted-in.`);
|
||||
return 0;
|
||||
}
|
||||
|
||||
const endpointData = await this.fetchEndpointData(executeFrom, executeTo);
|
||||
|
||||
/** STAGE 1 - Fetch Endpoint Agent Metrics
|
||||
*
|
||||
* Reads Endpoint Agent metrics out of the `.ds-metrics-endpoint.metrics` data stream
|
||||
* and buckets them by Endpoint Agent id and sorts by the top hit. The EP agent will
|
||||
* report its metrics once per day OR every time a policy change has occured. If
|
||||
* a metric document(s) exists for an EP agent we map to fleet agent and policy
|
||||
*/
|
||||
if (endpointData.endpointMetrics === undefined) {
|
||||
this.logger.debug(`no endpoint metrics to report`);
|
||||
return 0;
|
||||
}
|
||||
|
||||
const { body: endpointMetricsResponse } = endpointData.endpointMetrics as unknown as {
|
||||
body: EndpointMetricsAggregation;
|
||||
};
|
||||
|
||||
if (endpointMetricsResponse.aggregations === undefined) {
|
||||
this.logger.debug(`no endpoint metrics to report`);
|
||||
return 0;
|
||||
}
|
||||
|
||||
const endpointMetrics = endpointMetricsResponse.aggregations.endpoint_agents.buckets.map(
|
||||
(epMetrics) => {
|
||||
return {
|
||||
endpoint_agent: epMetrics.latest_metrics.hits.hits[0]._source.agent.id,
|
||||
endpoint_version: epMetrics.latest_metrics.hits.hits[0]._source.agent.version,
|
||||
endpoint_metrics: epMetrics.latest_metrics.hits.hits[0]._source,
|
||||
};
|
||||
}
|
||||
);
|
||||
|
||||
/** STAGE 2 - Fetch Fleet Agent Config
|
||||
*
|
||||
* As the policy id + policy version does not exist on the Endpoint Metrics document
|
||||
* we need to fetch information about the Fleet Agent and sync the metrics document
|
||||
* with the Agent's policy data.
|
||||
*
|
||||
*/
|
||||
const agentsResponse = endpointData.fleetAgentsResponse;
|
||||
|
||||
if (agentsResponse === undefined) {
|
||||
this.logger.debug('no fleet agent information available');
|
||||
return 0;
|
||||
}
|
||||
|
||||
const fleetAgents = agentsResponse.agents.reduce((cache, agent) => {
|
||||
if (agent.id === DefaultEndpointPolicyIdToIgnore) {
|
||||
return cache;
|
||||
}
|
||||
|
||||
if (agent.policy_id !== null && agent.policy_id !== undefined) {
|
||||
cache.set(agent.id, agent.policy_id);
|
||||
}
|
||||
|
||||
return cache;
|
||||
}, new Map<string, string>());
|
||||
|
||||
const endpointPolicyCache = new Map<string, PolicyData>();
|
||||
for (const policyInfo of fleetAgents.values()) {
|
||||
if (policyInfo !== null && policyInfo !== undefined && !endpointPolicyCache.has(policyInfo)) {
|
||||
const agentPolicy = await this.receiver.fetchPolicyConfigs(policyInfo);
|
||||
const packagePolicies = agentPolicy?.package_policies;
|
||||
|
||||
if (packagePolicies !== undefined && isPackagePolicyList(packagePolicies)) {
|
||||
packagePolicies
|
||||
.map((pPolicy) => pPolicy as PolicyData)
|
||||
.forEach((pPolicy) => {
|
||||
if (pPolicy.inputs[0].config !== undefined) {
|
||||
pPolicy.inputs.forEach((input) => {
|
||||
if (
|
||||
input.type === FLEET_ENDPOINT_PACKAGE &&
|
||||
input.config !== undefined &&
|
||||
policyInfo !== undefined
|
||||
) {
|
||||
endpointPolicyCache.set(policyInfo, pPolicy);
|
||||
}
|
||||
});
|
||||
}
|
||||
});
|
||||
}
|
||||
);
|
||||
|
||||
/** STAGE 2 - Fetch Fleet Agent Config
|
||||
*
|
||||
* As the policy id + policy version does not exist on the Endpoint Metrics document
|
||||
* we need to fetch information about the Fleet Agent and sync the metrics document
|
||||
* with the Agent's policy data.
|
||||
*
|
||||
*/
|
||||
const agentsResponse = endpointData.fleetAgentsResponse;
|
||||
|
||||
if (agentsResponse === undefined) {
|
||||
logger.debug('no fleet agent information available');
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
/** STAGE 3 - Fetch Endpoint Policy Responses
|
||||
*
|
||||
* Reads Endpoint Agent policy responses out of the `.ds-metrics-endpoint.policy*` data
|
||||
* stream and creates a local K/V structure that stores the policy response (V) with
|
||||
* the Endpoint Agent Id (K). A value will only exist if there has been a endpoint
|
||||
* enrolled in the last 24 hours OR a policy change has occurred. We only send
|
||||
* non-successful responses. If the field is null, we assume no responses in
|
||||
* the last 24h or no failures/warnings in the policy applied.
|
||||
*
|
||||
*/
|
||||
const { body: failedPolicyResponses } = endpointData.epPolicyResponse as unknown as {
|
||||
body: EndpointPolicyResponseAggregation;
|
||||
};
|
||||
const fleetAgents = agentsResponse.agents.reduce((cache, agent) => {
|
||||
if (agent.id === DefaultEndpointPolicyIdToIgnore) {
|
||||
return cache;
|
||||
}
|
||||
|
||||
// If there is no policy responses in the 24h > now then we will continue
|
||||
const policyResponses = failedPolicyResponses.aggregations
|
||||
? failedPolicyResponses.aggregations.policy_responses.buckets.reduce(
|
||||
(cache, endpointAgentId) => {
|
||||
const doc = endpointAgentId.latest_response.hits.hits[0];
|
||||
cache.set(endpointAgentId.key, doc);
|
||||
return cache;
|
||||
},
|
||||
new Map<string, EndpointPolicyResponseDocument>()
|
||||
)
|
||||
: new Map<string, EndpointPolicyResponseDocument>();
|
||||
if (agent.policy_id !== null && agent.policy_id !== undefined) {
|
||||
cache.set(agent.id, agent.policy_id);
|
||||
}
|
||||
|
||||
/** STAGE 4 - Create the telemetry log records
|
||||
*
|
||||
* Iterates through the endpoint metrics documents at STAGE 1 and joins them together
|
||||
* to form the telemetry log that is sent back to Elastic Security developers to
|
||||
* make improvements to the product.
|
||||
*
|
||||
*/
|
||||
try {
|
||||
const telemetryPayloads = endpointMetrics.map((endpoint) => {
|
||||
let policyConfig = null;
|
||||
let failedPolicy = null;
|
||||
return cache;
|
||||
}, new Map<string, string>());
|
||||
|
||||
const fleetAgentId = endpoint.endpoint_metrics.elastic.agent.id;
|
||||
const endpointAgentId = endpoint.endpoint_agent;
|
||||
const endpointPolicyCache = new Map<string, PolicyData>();
|
||||
for (const policyInfo of fleetAgents.values()) {
|
||||
if (
|
||||
policyInfo !== null &&
|
||||
policyInfo !== undefined &&
|
||||
!endpointPolicyCache.has(policyInfo)
|
||||
) {
|
||||
const agentPolicy = await receiver.fetchPolicyConfigs(policyInfo);
|
||||
const packagePolicies = agentPolicy?.package_policies;
|
||||
|
||||
const policyInformation = fleetAgents.get(fleetAgentId);
|
||||
if (policyInformation) {
|
||||
policyConfig = endpointPolicyCache.get(policyInformation) || null;
|
||||
|
||||
if (policyConfig) {
|
||||
failedPolicy = policyResponses.get(endpointAgentId);
|
||||
if (packagePolicies !== undefined && isPackagePolicyList(packagePolicies)) {
|
||||
packagePolicies
|
||||
.map((pPolicy) => pPolicy as PolicyData)
|
||||
.forEach((pPolicy) => {
|
||||
if (pPolicy.inputs[0].config !== undefined) {
|
||||
pPolicy.inputs.forEach((input) => {
|
||||
if (
|
||||
input.type === FLEET_ENDPOINT_PACKAGE &&
|
||||
input.config !== undefined &&
|
||||
policyInfo !== undefined
|
||||
) {
|
||||
endpointPolicyCache.set(policyInfo, pPolicy);
|
||||
}
|
||||
});
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const { cpu, memory, uptime } = endpoint.endpoint_metrics.Endpoint.metrics;
|
||||
|
||||
return {
|
||||
'@timestamp': executeTo,
|
||||
endpoint_id: endpointAgentId,
|
||||
endpoint_version: endpoint.endpoint_version,
|
||||
endpoint_package_version: policyConfig?.package?.version || null,
|
||||
endpoint_metrics: {
|
||||
cpu: cpu.endpoint,
|
||||
memory: memory.endpoint.private,
|
||||
uptime,
|
||||
},
|
||||
endpoint_meta: {
|
||||
os: endpoint.endpoint_metrics.host.os,
|
||||
},
|
||||
policy_config: policyConfig !== null ? policyConfig?.inputs[0].config.policy : {},
|
||||
policy_response:
|
||||
failedPolicy !== null && failedPolicy !== undefined
|
||||
? {
|
||||
agent_policy_status: failedPolicy._source.event.agent_id_status,
|
||||
manifest_version:
|
||||
failedPolicy._source.Endpoint.policy.applied.artifacts.global.version,
|
||||
status: failedPolicy._source.Endpoint.policy.applied.status,
|
||||
actions: failedPolicy._source.Endpoint.policy.applied.actions
|
||||
.map((action) => (action.status !== 'success' ? action : null))
|
||||
.filter((action) => action !== null),
|
||||
}
|
||||
: {},
|
||||
telemetry_meta: {
|
||||
metrics_timestamp: endpoint.endpoint_metrics['@timestamp'],
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
/**
|
||||
* STAGE 5 - Send the documents
|
||||
/** STAGE 3 - Fetch Endpoint Policy Responses
|
||||
*
|
||||
* Reads Endpoint Agent policy responses out of the `.ds-metrics-endpoint.policy*` data
|
||||
* stream and creates a local K/V structure that stores the policy response (V) with
|
||||
* the Endpoint Agent Id (K). A value will only exist if there has been a endpoint
|
||||
* enrolled in the last 24 hours OR a policy change has occurred. We only send
|
||||
* non-successful responses. If the field is null, we assume no responses in
|
||||
* the last 24h or no failures/warnings in the policy applied.
|
||||
*
|
||||
* Send the documents in a batches of 100
|
||||
*/
|
||||
batchTelemetryRecords(telemetryPayloads, 100).forEach((telemetryBatch) =>
|
||||
this.sender.sendOnDemand(TELEMETRY_CHANNEL_ENDPOINT_META, telemetryBatch)
|
||||
);
|
||||
return telemetryPayloads.length;
|
||||
} catch (err) {
|
||||
this.logger.warn('could not complete endpoint alert telemetry task');
|
||||
return 0;
|
||||
}
|
||||
const { body: failedPolicyResponses } = endpointData.epPolicyResponse as unknown as {
|
||||
body: EndpointPolicyResponseAggregation;
|
||||
};
|
||||
|
||||
// If there is no policy responses in the 24h > now then we will continue
|
||||
const policyResponses = failedPolicyResponses.aggregations
|
||||
? failedPolicyResponses.aggregations.policy_responses.buckets.reduce(
|
||||
(cache, endpointAgentId) => {
|
||||
const doc = endpointAgentId.latest_response.hits.hits[0];
|
||||
cache.set(endpointAgentId.key, doc);
|
||||
return cache;
|
||||
},
|
||||
new Map<string, EndpointPolicyResponseDocument>()
|
||||
)
|
||||
: new Map<string, EndpointPolicyResponseDocument>();
|
||||
|
||||
/** STAGE 4 - Create the telemetry log records
|
||||
*
|
||||
* Iterates through the endpoint metrics documents at STAGE 1 and joins them together
|
||||
* to form the telemetry log that is sent back to Elastic Security developers to
|
||||
* make improvements to the product.
|
||||
*
|
||||
*/
|
||||
try {
|
||||
const telemetryPayloads = endpointMetrics.map((endpoint) => {
|
||||
let policyConfig = null;
|
||||
let failedPolicy = null;
|
||||
|
||||
const fleetAgentId = endpoint.endpoint_metrics.elastic.agent.id;
|
||||
const endpointAgentId = endpoint.endpoint_agent;
|
||||
|
||||
const policyInformation = fleetAgents.get(fleetAgentId);
|
||||
if (policyInformation) {
|
||||
policyConfig = endpointPolicyCache.get(policyInformation) || null;
|
||||
|
||||
if (policyConfig) {
|
||||
failedPolicy = policyResponses.get(endpointAgentId);
|
||||
}
|
||||
}
|
||||
|
||||
const { cpu, memory, uptime } = endpoint.endpoint_metrics.Endpoint.metrics;
|
||||
|
||||
return {
|
||||
'@timestamp': taskExecutionPeriod.current,
|
||||
endpoint_id: endpointAgentId,
|
||||
endpoint_version: endpoint.endpoint_version,
|
||||
endpoint_package_version: policyConfig?.package?.version || null,
|
||||
endpoint_metrics: {
|
||||
cpu: cpu.endpoint,
|
||||
memory: memory.endpoint.private,
|
||||
uptime,
|
||||
},
|
||||
endpoint_meta: {
|
||||
os: endpoint.endpoint_metrics.host.os,
|
||||
},
|
||||
policy_config: policyConfig !== null ? policyConfig?.inputs[0].config.policy : {},
|
||||
policy_response:
|
||||
failedPolicy !== null && failedPolicy !== undefined
|
||||
? {
|
||||
agent_policy_status: failedPolicy._source.event.agent_id_status,
|
||||
manifest_version:
|
||||
failedPolicy._source.Endpoint.policy.applied.artifacts.global.version,
|
||||
status: failedPolicy._source.Endpoint.policy.applied.status,
|
||||
actions: failedPolicy._source.Endpoint.policy.applied.actions
|
||||
.map((action) => (action.status !== 'success' ? action : null))
|
||||
.filter((action) => action !== null),
|
||||
}
|
||||
: {},
|
||||
telemetry_meta: {
|
||||
metrics_timestamp: endpoint.endpoint_metrics['@timestamp'],
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
/**
|
||||
* STAGE 5 - Send the documents
|
||||
*
|
||||
* Send the documents in a batches of maxTelemetryBatch
|
||||
*/
|
||||
batchTelemetryRecords(telemetryPayloads, maxTelemetryBatch).forEach((telemetryBatch) =>
|
||||
sender.sendOnDemand(TELEMETRY_CHANNEL_ENDPOINT_META, telemetryBatch)
|
||||
);
|
||||
return telemetryPayloads.length;
|
||||
} catch (err) {
|
||||
logger.warn('could not complete endpoint alert telemetry task');
|
||||
return 0;
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function fetchEndpointData(
|
||||
receiver: TelemetryReceiver,
|
||||
executeFrom: string,
|
||||
executeTo: string
|
||||
) {
|
||||
const [fleetAgentsResponse, epMetricsResponse, policyResponse] = await Promise.allSettled([
|
||||
receiver.fetchFleetAgents(),
|
||||
receiver.fetchEndpointMetrics(executeFrom, executeTo),
|
||||
receiver.fetchEndpointPolicyResponses(executeFrom, executeTo),
|
||||
]);
|
||||
|
||||
return {
|
||||
fleetAgentsResponse:
|
||||
fleetAgentsResponse.status === 'fulfilled'
|
||||
? fleetAgentsResponse.value
|
||||
: EmptyFleetAgentResponse,
|
||||
endpointMetrics: epMetricsResponse.status === 'fulfilled' ? epMetricsResponse.value : undefined,
|
||||
epPolicyResponse: policyResponse.status === 'fulfilled' ? policyResponse.value : undefined,
|
||||
};
|
||||
}
|
||||
|
|
|
@ -5,7 +5,22 @@
|
|||
* 2.0.
|
||||
*/
|
||||
|
||||
export { TelemetryDiagTask as DiagnosticTask } from './diagnostic';
|
||||
export { TelemetryEndpointTask as EndpointTask } from './endpoint';
|
||||
export { TelemetryExceptionListsTask as ExceptionListsTask } from './security_lists';
|
||||
export { TelemetryDetectionRulesTask as DetectionRulesTask } from './detection_rule';
|
||||
import { SecurityTelemetryTaskConfig } from '../task';
|
||||
import { createTelemetryDiagnosticsTaskConfig } from './diagnostic';
|
||||
import { createTelemetryEndpointTaskConfig } from './endpoint';
|
||||
import { createTelemetrySecurityListTaskConfig } from './security_lists';
|
||||
import { createTelemetryDetectionRuleListsTaskConfig } from './detection_rule';
|
||||
import {
|
||||
MAX_SECURITY_LIST_TELEMETRY_BATCH,
|
||||
MAX_ENDPOINT_TELEMETRY_BATCH,
|
||||
MAX_DETECTION_RULE_TELEMETRY_BATCH,
|
||||
} from '../constants';
|
||||
|
||||
export function createTelemetryTaskConfigs(): SecurityTelemetryTaskConfig[] {
|
||||
return [
|
||||
createTelemetryDiagnosticsTaskConfig(),
|
||||
createTelemetryEndpointTaskConfig(MAX_SECURITY_LIST_TELEMETRY_BATCH),
|
||||
createTelemetrySecurityListTaskConfig(MAX_ENDPOINT_TELEMETRY_BATCH),
|
||||
createTelemetryDetectionRuleListsTaskConfig(MAX_DETECTION_RULE_TELEMETRY_BATCH),
|
||||
];
|
||||
}
|
||||
|
|
|
@ -6,117 +6,41 @@
|
|||
*/
|
||||
|
||||
import { loggingSystemMock } from 'src/core/server/mocks';
|
||||
import { TaskStatus } from '../../../../../task_manager/server';
|
||||
import { taskManagerMock } from '../../../../../task_manager/server/mocks';
|
||||
import { TelemetryExceptionListsTask, TelemetrySecuityListsTaskConstants } from './security_lists';
|
||||
import { createTelemetrySecurityListTaskConfig } from './security_lists';
|
||||
import { createMockTelemetryEventsSender, createMockTelemetryReceiver } from '../__mocks__';
|
||||
import {
|
||||
createMockTelemetryEventsSender,
|
||||
MockExceptionListsTask,
|
||||
createMockTelemetryReceiver,
|
||||
} from '../mocks';
|
||||
ENDPOINT_LIST_ID,
|
||||
ENDPOINT_EVENT_FILTERS_LIST_ID,
|
||||
} from '@kbn/securitysolution-list-constants';
|
||||
|
||||
describe('test exception list telemetry task functionality', () => {
|
||||
describe('security list telemetry task test', () => {
|
||||
let logger: ReturnType<typeof loggingSystemMock.createLogger>;
|
||||
|
||||
beforeEach(() => {
|
||||
logger = loggingSystemMock.createLogger();
|
||||
});
|
||||
|
||||
test('the trusted apps task can register', () => {
|
||||
const telemetryTrustedAppsTask = new TelemetryExceptionListsTask(
|
||||
logger,
|
||||
taskManagerMock.createSetup(),
|
||||
createMockTelemetryEventsSender(true),
|
||||
createMockTelemetryReceiver()
|
||||
);
|
||||
|
||||
expect(telemetryTrustedAppsTask).toBeInstanceOf(TelemetryExceptionListsTask);
|
||||
});
|
||||
|
||||
test('the exception list task should be registered', () => {
|
||||
const mockTaskManager = taskManagerMock.createSetup();
|
||||
new TelemetryExceptionListsTask(
|
||||
logger,
|
||||
mockTaskManager,
|
||||
createMockTelemetryEventsSender(true),
|
||||
createMockTelemetryReceiver()
|
||||
);
|
||||
|
||||
expect(mockTaskManager.registerTaskDefinitions).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test('the exception list task should be scheduled', async () => {
|
||||
const mockTaskManagerSetup = taskManagerMock.createSetup();
|
||||
const telemetryTrustedAppsTask = new TelemetryExceptionListsTask(
|
||||
logger,
|
||||
mockTaskManagerSetup,
|
||||
createMockTelemetryEventsSender(true),
|
||||
createMockTelemetryReceiver()
|
||||
);
|
||||
|
||||
const mockTaskManagerStart = taskManagerMock.createStart();
|
||||
await telemetryTrustedAppsTask.start(mockTaskManagerStart);
|
||||
expect(mockTaskManagerStart.ensureScheduled).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test('the exception list task should not query elastic if telemetry is not opted in', async () => {
|
||||
const mockSender = createMockTelemetryEventsSender(false);
|
||||
const mockReceiver = createMockTelemetryReceiver();
|
||||
const mockTaskManager = taskManagerMock.createSetup();
|
||||
new MockExceptionListsTask(logger, mockTaskManager, mockSender, mockReceiver);
|
||||
|
||||
const mockTaskInstance = {
|
||||
id: TelemetrySecuityListsTaskConstants.TYPE,
|
||||
runAt: new Date(),
|
||||
attempts: 0,
|
||||
ownerId: '',
|
||||
status: TaskStatus.Running,
|
||||
startedAt: new Date(),
|
||||
scheduledAt: new Date(),
|
||||
retryAt: new Date(),
|
||||
params: {},
|
||||
state: {},
|
||||
taskType: TelemetrySecuityListsTaskConstants.TYPE,
|
||||
test('security list telemetry task should fetch security list data', async () => {
|
||||
const testTaskExecutionPeriod = {
|
||||
last: undefined,
|
||||
current: new Date().toISOString(),
|
||||
};
|
||||
const createTaskRunner =
|
||||
mockTaskManager.registerTaskDefinitions.mock.calls[0][0][
|
||||
TelemetrySecuityListsTaskConstants.TYPE
|
||||
].createTaskRunner;
|
||||
const taskRunner = createTaskRunner({ taskInstance: mockTaskInstance });
|
||||
await taskRunner.run();
|
||||
expect(mockReceiver.fetchTrustedApplications).not.toHaveBeenCalled();
|
||||
});
|
||||
const mockTelemetryEventsSender = createMockTelemetryEventsSender();
|
||||
const mockTelemetryReceiver = createMockTelemetryReceiver();
|
||||
const telemetrySecurityListTaskConfig = createTelemetrySecurityListTaskConfig(1);
|
||||
|
||||
test('the exception list task should query elastic if telemetry opted in', async () => {
|
||||
const mockSender = createMockTelemetryEventsSender(true);
|
||||
const mockTaskManager = taskManagerMock.createSetup();
|
||||
const mockReceiver = createMockTelemetryReceiver();
|
||||
const telemetryTrustedAppsTask = new MockExceptionListsTask(
|
||||
await telemetrySecurityListTaskConfig.runTask(
|
||||
'test-id',
|
||||
logger,
|
||||
mockTaskManager,
|
||||
mockSender,
|
||||
mockReceiver
|
||||
mockTelemetryReceiver,
|
||||
mockTelemetryEventsSender,
|
||||
testTaskExecutionPeriod
|
||||
);
|
||||
|
||||
const mockTaskInstance = {
|
||||
id: TelemetrySecuityListsTaskConstants.TYPE,
|
||||
runAt: new Date(),
|
||||
attempts: 0,
|
||||
ownerId: '',
|
||||
status: TaskStatus.Running,
|
||||
startedAt: new Date(),
|
||||
scheduledAt: new Date(),
|
||||
retryAt: new Date(),
|
||||
params: {},
|
||||
state: {},
|
||||
taskType: TelemetrySecuityListsTaskConstants.TYPE,
|
||||
};
|
||||
const createTaskRunner =
|
||||
mockTaskManager.registerTaskDefinitions.mock.calls[0][0][
|
||||
TelemetrySecuityListsTaskConstants.TYPE
|
||||
].createTaskRunner;
|
||||
const taskRunner = createTaskRunner({ taskInstance: mockTaskInstance });
|
||||
await taskRunner.run();
|
||||
expect(telemetryTrustedAppsTask.runTask).toHaveBeenCalled();
|
||||
expect(mockTelemetryReceiver.fetchTrustedApplications).toHaveBeenCalled();
|
||||
expect(mockTelemetryReceiver.fetchEndpointList).toHaveBeenCalledWith(ENDPOINT_LIST_ID);
|
||||
expect(mockTelemetryReceiver.fetchEndpointList).toHaveBeenCalledWith(
|
||||
ENDPOINT_EVENT_FILTERS_LIST_ID
|
||||
);
|
||||
});
|
||||
});
|
||||
|
|
|
@ -5,17 +5,11 @@
|
|||
* 2.0.
|
||||
*/
|
||||
|
||||
import moment from 'moment';
|
||||
import { Logger } from 'src/core/server';
|
||||
import {
|
||||
ENDPOINT_LIST_ID,
|
||||
ENDPOINT_EVENT_FILTERS_LIST_ID,
|
||||
} from '@kbn/securitysolution-list-constants';
|
||||
import {
|
||||
ConcreteTaskInstance,
|
||||
TaskManagerSetupContract,
|
||||
TaskManagerStartContract,
|
||||
} from '../../../../../task_manager/server';
|
||||
import {
|
||||
LIST_ENDPOINT_EXCEPTION,
|
||||
LIST_ENDPOINT_EVENT_FILTER,
|
||||
|
@ -25,119 +19,64 @@ import {
|
|||
import { batchTelemetryRecords, templateExceptionList } from '../helpers';
|
||||
import { TelemetryEventsSender } from '../sender';
|
||||
import { TelemetryReceiver } from '../receiver';
|
||||
import { TaskExecutionPeriod } from '../task';
|
||||
|
||||
export const TelemetrySecuityListsTaskConstants = {
|
||||
TIMEOUT: '3m',
|
||||
TYPE: 'security:telemetry-lists',
|
||||
INTERVAL: '24h',
|
||||
VERSION: '1.0.0',
|
||||
};
|
||||
export function createTelemetrySecurityListTaskConfig(maxTelemetryBatch: number) {
|
||||
return {
|
||||
type: 'security:telemetry-lists',
|
||||
title: 'Security Solution Lists Telemetry',
|
||||
interval: '24h',
|
||||
timeout: '3m',
|
||||
version: '1.0.0',
|
||||
runTask: async (
|
||||
taskId: string,
|
||||
logger: Logger,
|
||||
receiver: TelemetryReceiver,
|
||||
sender: TelemetryEventsSender,
|
||||
taskExecutionPeriod: TaskExecutionPeriod
|
||||
) => {
|
||||
let count = 0;
|
||||
|
||||
const MAX_TELEMETRY_BATCH = 1_000;
|
||||
// Lists Telemetry: Trusted Applications
|
||||
|
||||
export class TelemetryExceptionListsTask {
|
||||
private readonly logger: Logger;
|
||||
private readonly sender: TelemetryEventsSender;
|
||||
private readonly receiver: TelemetryReceiver;
|
||||
const trustedApps = await receiver.fetchTrustedApplications();
|
||||
if (trustedApps?.data) {
|
||||
const trustedAppsJson = templateExceptionList(trustedApps.data, LIST_TRUSTED_APPLICATION);
|
||||
logger.debug(`Trusted Apps: ${trustedAppsJson}`);
|
||||
count += trustedAppsJson.length;
|
||||
|
||||
constructor(
|
||||
logger: Logger,
|
||||
taskManager: TaskManagerSetupContract,
|
||||
sender: TelemetryEventsSender,
|
||||
receiver: TelemetryReceiver
|
||||
) {
|
||||
this.logger = logger;
|
||||
this.sender = sender;
|
||||
this.receiver = receiver;
|
||||
batchTelemetryRecords(trustedAppsJson, maxTelemetryBatch).forEach((batch) =>
|
||||
sender.sendOnDemand(TELEMETRY_CHANNEL_LISTS, batch)
|
||||
);
|
||||
}
|
||||
|
||||
taskManager.registerTaskDefinitions({
|
||||
[TelemetrySecuityListsTaskConstants.TYPE]: {
|
||||
title: 'Security Solution Lists Telemetry',
|
||||
timeout: TelemetrySecuityListsTaskConstants.TIMEOUT,
|
||||
createTaskRunner: ({ taskInstance }: { taskInstance: ConcreteTaskInstance }) => {
|
||||
const { state } = taskInstance;
|
||||
// Lists Telemetry: Endpoint Exceptions
|
||||
|
||||
return {
|
||||
run: async () => {
|
||||
const taskExecutionTime = moment().utc().toISOString();
|
||||
const hits = await this.runTask(taskInstance.id);
|
||||
const epExceptions = await receiver.fetchEndpointList(ENDPOINT_LIST_ID);
|
||||
if (epExceptions?.data) {
|
||||
const epExceptionsJson = templateExceptionList(epExceptions.data, LIST_ENDPOINT_EXCEPTION);
|
||||
logger.debug(`EP Exceptions: ${epExceptionsJson}`);
|
||||
count += epExceptionsJson.length;
|
||||
|
||||
return {
|
||||
state: {
|
||||
lastExecutionTimestamp: taskExecutionTime,
|
||||
runs: (state.runs || 0) + 1,
|
||||
hits,
|
||||
},
|
||||
};
|
||||
},
|
||||
cancel: async () => {},
|
||||
};
|
||||
},
|
||||
},
|
||||
});
|
||||
}
|
||||
batchTelemetryRecords(epExceptionsJson, maxTelemetryBatch).forEach((batch) =>
|
||||
sender.sendOnDemand(TELEMETRY_CHANNEL_LISTS, batch)
|
||||
);
|
||||
}
|
||||
|
||||
public start = async (taskManager: TaskManagerStartContract) => {
|
||||
try {
|
||||
await taskManager.ensureScheduled({
|
||||
id: this.getTaskId(),
|
||||
taskType: TelemetrySecuityListsTaskConstants.TYPE,
|
||||
scope: ['securitySolution'],
|
||||
schedule: {
|
||||
interval: TelemetrySecuityListsTaskConstants.INTERVAL,
|
||||
},
|
||||
state: { runs: 0 },
|
||||
params: { version: TelemetrySecuityListsTaskConstants.VERSION },
|
||||
});
|
||||
} catch (e) {
|
||||
this.logger.error(`Error scheduling task, received ${e.message}`);
|
||||
}
|
||||
};
|
||||
// Lists Telemetry: Endpoint Event Filters
|
||||
|
||||
private getTaskId = (): string => {
|
||||
return `${TelemetrySecuityListsTaskConstants.TYPE}:${TelemetrySecuityListsTaskConstants.VERSION}`;
|
||||
};
|
||||
const epFilters = await receiver.fetchEndpointList(ENDPOINT_EVENT_FILTERS_LIST_ID);
|
||||
if (epFilters?.data) {
|
||||
const epFiltersJson = templateExceptionList(epFilters.data, LIST_ENDPOINT_EVENT_FILTER);
|
||||
logger.debug(`EP Event Filters: ${epFiltersJson}`);
|
||||
count += epFiltersJson.length;
|
||||
|
||||
public runTask = async (taskId: string) => {
|
||||
if (taskId !== this.getTaskId()) {
|
||||
return 0;
|
||||
}
|
||||
batchTelemetryRecords(epFiltersJson, maxTelemetryBatch).forEach((batch) =>
|
||||
sender.sendOnDemand(TELEMETRY_CHANNEL_LISTS, batch)
|
||||
);
|
||||
}
|
||||
|
||||
const isOptedIn = await this.sender.isTelemetryOptedIn();
|
||||
if (!isOptedIn) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
// Lists Telemetry: Trusted Applications
|
||||
|
||||
const trustedApps = await this.receiver.fetchTrustedApplications();
|
||||
const trustedAppsJson = templateExceptionList(trustedApps.data, LIST_TRUSTED_APPLICATION);
|
||||
this.logger.debug(`Trusted Apps: ${trustedAppsJson}`);
|
||||
|
||||
batchTelemetryRecords(trustedAppsJson, MAX_TELEMETRY_BATCH).forEach((batch) =>
|
||||
this.sender.sendOnDemand(TELEMETRY_CHANNEL_LISTS, batch)
|
||||
);
|
||||
|
||||
// Lists Telemetry: Endpoint Exceptions
|
||||
|
||||
const epExceptions = await this.receiver.fetchEndpointList(ENDPOINT_LIST_ID);
|
||||
const epExceptionsJson = templateExceptionList(epExceptions.data, LIST_ENDPOINT_EXCEPTION);
|
||||
this.logger.debug(`EP Exceptions: ${epExceptionsJson}`);
|
||||
|
||||
batchTelemetryRecords(epExceptionsJson, MAX_TELEMETRY_BATCH).forEach((batch) =>
|
||||
this.sender.sendOnDemand(TELEMETRY_CHANNEL_LISTS, batch)
|
||||
);
|
||||
|
||||
// Lists Telemetry: Endpoint Event Filters
|
||||
|
||||
const epFilters = await this.receiver.fetchEndpointList(ENDPOINT_EVENT_FILTERS_LIST_ID);
|
||||
const epFiltersJson = templateExceptionList(epFilters.data, LIST_ENDPOINT_EVENT_FILTER);
|
||||
this.logger.debug(`EP Event Filters: ${epFiltersJson}`);
|
||||
|
||||
batchTelemetryRecords(epFiltersJson, MAX_TELEMETRY_BATCH).forEach((batch) =>
|
||||
this.sender.sendOnDemand(TELEMETRY_CHANNEL_LISTS, batch)
|
||||
);
|
||||
|
||||
return trustedAppsJson.length + epExceptionsJson.length + epFiltersJson.length;
|
||||
return count;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
|
Loading…
Reference in a new issue