import logging from defence360agent.contracts.config import ( _DOS_DETECTOR_MIN_LIMIT, ConfigFile, ) logger = logging.getLogger(__name__) def migrate(migrator, database, fake=False, **kwargs): if fake: return try: config_file = ConfigFile() config = config_file.config_to_dict(False) if "DOS" not in config: return if "max_connections" in config["DOS"] and isinstance( config["DOS"]["max_connections"], int ): config["DOS"]["default_limit"] = max( config["DOS"]["max_connections"], _DOS_DETECTOR_MIN_LIMIT ) del config["DOS"]["max_connections"] if "timeout" in config["DOS"]: config["DOS"]["interval"] = config["DOS"]["timeout"] del config["DOS"]["timeout"] config_file.dict_to_config(config, overwrite=True) except Exception: logger.exception("Failed to replace DOS settings") def rollback(migrator, database, fake=False, **kwargs): pass