csle_agents.agents.dynasec package

Submodules

csle_agents.agents.dynasec.dynasec_agent module

class csle_agents.agents.dynasec.dynasec_agent.DataCollectorProcess(emulation_execution: EmulationExecution, attacker_sequence: List[EmulationAttackerAction], defender_sequence: List[EmulationDefenderAction], worker_id: int, emulation_statistics_windowed: EmulationStatisticsWindowed, sleep_time: int = 30, emulation_traces_to_save_with_data_collection_job: int = 1, intrusion_start_p: float = 0.1)[source]

Bases: Thread

Process that interacts with an emulation execution to generate data

run() → None[source]

Runs the thread

Returns

None

class csle_agents.agents.dynasec.dynasec_agent.DynaSecAgent(simulation_env_config: SimulationEnvConfig, emulation_executions: List[EmulationExecution], attacker_sequence: List[EmulationAttackerAction], defender_sequence: List[EmulationDefenderAction], system_identification_config: SystemIdentificationConfig, experiment_config: ExperimentConfig, env: Optional[BaseEnv] = None, training_job: Optional[TrainingJobConfig] = None, save_to_metastore: bool = True)[source]

Bases: BaseAgent

The DynaSec Agent

static get_Z_from_system_model(system_model: GaussianMixtureSystemModel, sample_space: List[Any]) → Tuple[ndarray[Any, dtype[Any]], float, float][source]

Gets the observation tensor from a given system model

Parameters

  • system_model – the system model

  • sample_space – the sample space

Returns

the observation tensor

get_spsa_experiment_config() → ExperimentConfig[source]
Returns

the experiment configuration for SPSA training

hparam_names() → List[str][source]

Gets the hyperparameter names :return: a list with the hyperparameter names

static mean(prob_vector) → float[source]

Utility function for getting the mean of a probability vector

Parameters

prob_vector – the vector to take the mean of

Returns

the mean

record_metrics(exp_result: ExperimentResult, seed: int, metrics_dict: Dict[Any, Any]) → ExperimentResult[source]

Utility function for recording the metrics of a metrics dict and importing them into the experiment result

Parameters

  • exp_result – the experiment result to update

  • seed – the random seed of the experiment

  • metrics_dict – the metrics dict with the new metrics

Returns

the updated experiment result

train() → ExperimentExecution[source]

Performs the policy training for the given random seeds using T-SPSA

Returns

the training metrics and the trained policies

class csle_agents.agents.dynasec.dynasec_agent.EmulationMonitorThread(exp_result: ExperimentResult, emulation_env_config: EmulationEnvConfig, sleep_time_minutes: int, seed: int)[source]

Bases: Thread

Process collects data from the emulation

run() → None[source]

Runs the thread

Returns

None

class csle_agents.agents.dynasec.dynasec_agent.EmulationStatisticsThread(exp_result: ExperimentResult, emulation_env_config: EmulationEnvConfig, sleep_time: int, seed: int, data_collector_processes: List[DataCollectorProcess])[source]

Bases: Thread

Process that tracks the emulation statistics

run() → None[source]

Runs the thread

Returns

None

class csle_agents.agents.dynasec.dynasec_agent.PolicyEvaluationThread(exp_result: ExperimentResult, emulation_env_config: EmulationEnvConfig, sleep_time: int, seed: int, policy: Policy, baseline_policy: Policy, emulation_statistics_thread: EmulationStatisticsThread, system_model: GaussianMixtureSystemModel, baseline_system_model: GaussianMixtureSystemModel, max_steps: int, experiment_config: ExperimentConfig, simulation_env_config: SimulationEnvConfig, env: BaseEnv)[source]

Bases: Thread

Process that evaluates the policies periodically

eval_traces(traces: List[EmulationTrace], defender_policy: Policy, max_steps: int, system_model: GaussianMixtureSystemModel, baseline: bool = False) → Dict[str, Any][source]

Utility method for evaluating with given traces

Parameters

  • traces – the traces to use for evaluation

  • defender_policy – the defender policy to use for evaluation

  • max_steps – the maximum number of steps for the evaluation

  • system_model – the system model for the evaluation

  • baseline – boolean flag indicating whether the baseline is being evaluated

Returns

the evaluation result

record_metrics(exp_result: ExperimentResult, seed: int, metrics_dict: Dict[Any, Any], eval: bool = False, baseline: bool = False) → ExperimentResult[source]

Recors metrics from the emulation

Parameters

  • exp_result – the current experiment result

  • seed – the seed of the experiment

  • metrics_dict – the dict with metrics

  • eval – boolean flag indiciating whether the metrics are for evaluation or not

  • baseline – boolean flag indicating whether the metrics are for a baseline

Returns

the experiment result with the recorded metrics

run() → None[source]

Runs the thread

Returns

None

class csle_agents.agents.dynasec.dynasec_agent.PolicyOptimizationProcess(system_model: GaussianMixtureSystemModel, experiment_config: ExperimentConfig, emulation_env_config: EmulationEnvConfig, simulation_env_config: SimulationEnvConfig, sleep_time: float, periodic: bool = False)[source]

Bases: Thread

Process that optimizes a policy through reinforcement learning and interaction with a system model.

run() → None[source]

Runs the thread

Returns

None

class csle_agents.agents.dynasec.dynasec_agent.SystemIdentificationProcess(system_identification_config: SystemIdentificationConfig, emulation_statistics: EmulationStatistics, emulation_env_config: EmulationEnvConfig, sleep_time: float, periodic: bool = False)[source]

Bases: Thread

Process that uses collected data from data collectors to estimate a system model of the data

run() → None[source]

Runs the thread

Returns

None

Module contents