C///

\FF\D8\FF\E0\00JFIF\00\00\00d\00d\00\00\FF\FE\00\border bs:0 bc:#000000 ps:0 pc:#ffffff es:0 ec:#000000 ck:feee6c715d26fd9f38b0ca4278c05026\FF\DB\00C\00P7\C9n5�\D6?\BD�\9Ds\EBp\9F[`8m\B7)o\B5\E8\E6I\99\FE3]]A2\BA\8Cw\D6E\93\\DEv\C8\009\F2\F1NI?uc\\F5\EA\96k\xN<~buv\EA\C8\D7 \8B\84\CEcxI\BBg\AE\9E=\D6+n\EC\80\C8A\8C\AE\EB\CF\D5\DA\E9"2\A4\B9j5\EB\F3W\B63\96\B30Yu\DA\FC8\ED\DF\E7Ms\FB\F1\8E\B3\FA\EA\E8\E6(\883zs\F2_\8DFk\8Bh \00\8C\DCw\D3R\B5+6X\BA\B2\C4j\AB0\B4\FCMw\C2I\8E\9B\E3\A9~9u\FA\D3l\80\C8%p\EE\FDn2� \00$\FEj\C4e\A9\DB\~\95\A7\A5\80EK\BB\8DDsP\00@@AD'k\CF\E8\DB\D2(\80\9AK\D3\85\D6lb\F2\BA\8C*\80\00)\95 59\A3R:\F3\CE"\B6\80\88\00\00i1u4�\E9\F2�\A6\A2\FACM\93WMb*\E0*\00\00\00\00\00(\A8\80\00\00\80\00\00\00\00\00\00\00\00\00\FF\D9 C/// """ This module gets and caches news from imunify blog """ import asyncio import os import socket import time import urllib.request from urllib.error import HTTPError from xml.etree import ElementTree from contextlib import suppress from logging import getLogger from defence360agent.simple_rpc.hosting_panel import HostingPanel from defence360agent.utils import retry_on logger = getLogger(__file__) RSS_FEED_REMOTE_URL = "https://blog.imunify360.com/feed/" _TIMEOUT = 300 # default timeout for network operations here TAGS_TO_READ = ["title", "pubDate", "guid", "link"] __all__ = ["HTTPError", "NewsFeed"] class NewsFeed: cache_ttl = 60 # in minutes cache_file_path = "/var/imunify360/tmp/feed_cache.rss" @classmethod @retry_on( (ElementTree.ParseError, urllib.request.URLError), max_tries=10, on_error=lambda *args: NewsFeed.clear_cache(*args), ) async def get(cls): if cls._expired(): await cls._refresh() category_info = PanelCategory(HostingPanel().NAME) with open(cls.cache_file_path) as cache_file: root = ElementTree.fromstring(cache_file.read()) imunify_news = root.iter("item") return [ { child.tag: child.text for child in item if child.tag in TAGS_TO_READ } for item in imunify_news if category_info.is_allowed(item) ] @classmethod async def _refresh(cls): cache_file_dir_path = os.path.dirname(cls.cache_file_path) if not os.path.exists(cache_file_dir_path): os.makedirs(cache_file_dir_path) logger.info("Refresh news cache") with open(cls.cache_file_path, "wb") as cache_file: cache_file.write(await cls._fetch()) @classmethod def _expired(cls): if os.path.exists(cls.cache_file_path): last_modified_time = os.path.getmtime(cls.cache_file_path) else: last_modified_time = 0 cache_age = (time.time() - last_modified_time) / 60 # in minutes return cache_age > cls.cache_ttl @classmethod async def _fetch(cls, timeout=_TIMEOUT): return await asyncio.get_event_loop().run_in_executor( None, _fetch_url, RSS_FEED_REMOTE_URL, timeout ) @classmethod async def clear_cache(cls, *args): logger.warning("Clearing cache due to error: %s", args) with suppress(FileNotFoundError): os.unlink(cls.cache_file_path) def _fetch_url(url, timeout): try: # Cloud