22

✏️ Editing: whitelist_cache.py

#!/opt/imunify360/venv/bin/python

import asyncio

from defence360agent.contracts.config import Model
from defence360agent.internals import logger
from defence360agent.model import instance, tls_check
from im360.contracts.config import IPSET_LISTS_PATH

from im360.files import WHITELISTS, Index
from im360.model.cache_sources import (
    CountryIPListCacheSource,
    GlobalwhitelistCacheSource,
    WhitelistCacheSourceFromCSF,
    WhitelistCacheSourceFromSystemSettings,
)
from im360.model.firewall import IPList
from im360.subsys import csf

async def print_whitelist_cache_sources():
    csf_files = csf.CSF_IMUNIFY_IPLISTS_MAPPING["WHITE"]
    for source in [
        CountryIPListCacheSource(IPList.WHITE),
        GlobalwhitelistCacheSource(),
        WhitelistCacheSourceFromSystemSettings(),
    ] + (
        [WhitelistCacheSourceFromCSF(*csf_files)]
        if await csf.is_running()
        else []
    ):
        for ip, _ in await source.fetch_all():
            print(ip)

def setup_environment():
    logger.reconfigure()
    instance.db.init(Model.PATH)
    instance.db.execute_sql("ATTACH ? AS resident", (Model.RESIDENT_PATH,))
    instance.db.execute_sql("ATTACH ? AS ipsetlists", (IPSET_LISTS_PATH,))
    Index.add_type(WHITELISTS, "whitelist/v2", 0o770, 0o660, all_zip=True)

def main():
    tls_check.reset()
    setup_environment()
    loop = asyncio.get_event_loop()
    loop.run_until_complete(print_whitelist_cache_sources())

if __name__ == "__main__":
    main()

MR SKB