csle_agents.agents.dynasec package

Submodules

csle_agents.agents.dynasec.dynasec_agent module

class csle_agents.agents.dynasec.dynasec_agent.DataCollectorProcess(emulation_execution: csle_common.dao.emulation_config.emulation_execution.EmulationExecution, attacker_sequence: List[csle_common.dao.emulation_action.attacker.emulation_attacker_action.EmulationAttackerAction], defender_sequence: List[csle_common.dao.emulation_action.defender.emulation_defender_action.EmulationDefenderAction], worker_id: int, emulation_statistics_windowed: csle_common.dao.emulation_config.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: threading.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: csle_common.dao.simulation_config.simulation_env_config.SimulationEnvConfig, emulation_executions: List[csle_common.dao.emulation_config.emulation_execution.EmulationExecution], attacker_sequence: List[csle_common.dao.emulation_action.attacker.emulation_attacker_action.EmulationAttackerAction], defender_sequence: List[csle_common.dao.emulation_action.defender.emulation_defender_action.EmulationDefenderAction], system_identification_config: csle_common.dao.system_identification.system_identification_config.SystemIdentificationConfig, experiment_config: csle_common.dao.training.experiment_config.ExperimentConfig, env: Optional[csle_common.dao.simulation_config.base_env.BaseEnv] = None, training_job: Optional[csle_common.dao.jobs.training_job_config.TrainingJobConfig] = None, save_to_metastore: bool = True)[source]

Bases: csle_agents.agents.base.base_agent.BaseAgent

The DynaSec Agent

static get_Z_from_system_model(system_model: csle_common.dao.system_identification.gaussian_mixture_system_model.GaussianMixtureSystemModel, sample_space: List[Any]) → Tuple[numpy.ndarray[Any, numpy.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() → csle_common.dao.training.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: csle_common.dao.training.experiment_result.ExperimentResult, seed: int, metrics_dict: Dict[Any, Any]) → csle_common.dao.training.experiment_result.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() → csle_common.dao.training.experiment_execution.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: csle_common.dao.training.experiment_result.ExperimentResult, emulation_env_config: csle_common.dao.emulation_config.emulation_env_config.EmulationEnvConfig, sleep_time_minutes: int, seed: int)[source]

Bases: threading.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: csle_common.dao.training.experiment_result.ExperimentResult, emulation_env_config: csle_common.dao.emulation_config.emulation_env_config.EmulationEnvConfig, sleep_time: int, seed: int, data_collector_processes: List[csle_agents.agents.dynasec.dynasec_agent.DataCollectorProcess])[source]

Bases: threading.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: csle_common.dao.training.experiment_result.ExperimentResult, emulation_env_config: csle_common.dao.emulation_config.emulation_env_config.EmulationEnvConfig, sleep_time: int, seed: int, policy: csle_common.dao.training.policy.Policy, baseline_policy: csle_common.dao.training.policy.Policy, emulation_statistics_thread: csle_agents.agents.dynasec.dynasec_agent.EmulationStatisticsThread, system_model: csle_common.dao.system_identification.gaussian_mixture_system_model.GaussianMixtureSystemModel, baseline_system_model: csle_common.dao.system_identification.gaussian_mixture_system_model.GaussianMixtureSystemModel, max_steps: int, experiment_config: csle_common.dao.training.experiment_config.ExperimentConfig, simulation_env_config: csle_common.dao.simulation_config.simulation_env_config.SimulationEnvConfig, env: csle_common.dao.simulation_config.base_env.BaseEnv)[source]

Bases: threading.Thread

Process that evaluates the policies periodically

eval_traces(traces: List[csle_common.dao.emulation_config.emulation_trace.EmulationTrace], defender_policy: csle_common.dao.training.policy.Policy, max_steps: int, system_model: csle_common.dao.system_identification.gaussian_mixture_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: csle_common.dao.training.experiment_result.ExperimentResult, seed: int, metrics_dict: Dict[Any, Any], eval: bool = False, baseline: bool = False) → csle_common.dao.training.experiment_result.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: csle_common.dao.system_identification.gaussian_mixture_system_model.GaussianMixtureSystemModel, experiment_config: csle_common.dao.training.experiment_config.ExperimentConfig, emulation_env_config: csle_common.dao.emulation_config.emulation_env_config.EmulationEnvConfig, simulation_env_config: csle_common.dao.simulation_config.simulation_env_config.SimulationEnvConfig, sleep_time: float, periodic: bool = False)[source]

Bases: threading.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: csle_common.dao.system_identification.system_identification_config.SystemIdentificationConfig, emulation_statistics: csle_common.dao.system_identification.emulation_statistics.EmulationStatistics, emulation_env_config: csle_common.dao.emulation_config.emulation_env_config.EmulationEnvConfig, sleep_time: float, periodic: bool = False)[source]

Bases: threading.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