from defence360agent.contracts.config import IConfig, LocalConfig from defence360agent.utils import log_error_and_ignore @log_error_and_ignore() def migrate( migrator, database, fake=False, config_file: IConfig = LocalConfig(), **kwargs ): if fake: return config = config_file.config_to_dict() if not config: return # rename `max_incident_repetition` to `max_incidents` block_by_severity = config.setdefault("MOD_SEC_BLOCK_BY_SEVERITY", {}) value = block_by_severity.pop("max_incident_repetition", None) if value: block_by_severity["max_incidents"] = value custom_rule_list = config.setdefault("MOD_SEC_BLOCK_BY_CUSTOM_RULE", {}) for custom_rule_conf in custom_rule_list.values(): value = custom_rule_conf.pop("max_incident_repetition", None) if value: custom_rule_conf["max_incidents"] = value if config.get("INCIDENT_LIST"): # rename section `INCIDENT_LIST` to `INCIDENT_LOGGING` config["INCIDENT_LOGGING"] = config.pop("INCIDENT_LIST", {}) # move fields auto_cleanup_conf = config.pop("AUTOCLEANUP", None) if auto_cleanup_conf: config["INCIDENT_LOGGING"].update(**auto_cleanup_conf) config_file.dict_to_config(config, validate=False, overwrite=True) def rollback(migrator, database, fake=False, **kwargs): pass