def __init__()

in plugins/configuration.py [0:0]

• 32 lines of code
• 8 McCabe index (conditional complexity)

    def __init__(self, yml):
        self.database_filepath = yml.get("database", "blocky.sqlite")
        self.sqlite = asfpy.sqlite.DB(self.database_filepath)
        self.default_expire_seconds = yml.get("default_expire", DEFAULT_EXPIRE)
        self.index_pattern = yml.get("index_pattern", DEFAULT_INDEX_PATTERN)
        self.elasticsearch_url = yml.get("elasticsearch_url")
        self.elasticsearch = elasticsearch.AsyncElasticsearch(hosts=[self.elasticsearch_url])
        self.http_ip = yml.get("bind_ip", "127.0.0.1")
        self.http_port = int(yml.get("bind_port", 8080))
        self.client_iptables = {}  # Uploaded iptables from blocky clients. Only kept in memory.
        self.pubsub_host = yml.get('pubsub_host')
        self.pubsub_user = yml.get('pubsub_user')
        self.pubsub_password = yml.get('pubsub_password')

        # Create table if not there yet
        new_db = False
        if not self.sqlite.table_exists("rules"):
            print(f"Database file {self.database_filepath} is empty, initializing tables")
            self.sqlite.run(plugins.db_create.CREATE_DB_RULES)
            self.sqlite.run(plugins.db_create.CREATE_DB_LISTS)
            self.sqlite.run(plugins.db_create.CREATE_DB_AUDIT)
            print(f"Database file {self.database_filepath} has been successfully initialized")
            new_db = True

        # Init and fetch existing blocks and allows
        self.block_list = plugins.lists.List(self, "block")
        self.allow_list = plugins.lists.List(self, "allow")

        # Seed new DB with default allows if needed
        if new_db:
            for entry in DEFAULT_ALLOW_LIST:
                self.allow_list.add(
                    ip=entry,
                    timestamp=0,
                    expires=-1,
                    reason="Default allowed ranges (local network)",
                    host="*",
                )