#!/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()