ÿØÿàJFIFÿþ ÿÛC       ÿÛC ÿÀÿÄÿÄ"#QrÿÄÿÄ&1!A"2qQaáÿÚ ?Øy,æ/3JæݹÈ߲؋5êXw²±ÉyˆR”¾I0ó2—PI¾IÌÚiMö¯–þrìN&"KgX:Šíµ•nTJnLK„…@!‰-ý ùúmë;ºgµŒ&ó±hw’¯Õ@”Ü— 9ñ-ë.²1<yà‚¹ïQÐU„ہ?.’¦èûbß±©Ö«Âw*VŒ) `$‰bØԟ’ëXÖ-ËTÜíGÚ3ð«g Ÿ§¯—Jx„–’U/ÂÅv_s(Hÿ@TñJÑãõçn­‚!ÈgfbÓc­:él[ðQe 9ÀPLbÃãCµm[5¿ç'ªjglå‡Ûí_§Úõl-;"PkÞÞÁQâ¼_Ñ^¢SŸx?"¸¦ùY騐ÒOÈ q’`~~ÚtËU¹CڒêV  I1Áß_ÿÙ"""IPEchoAPI - returns real IP address of the host (behind NAT)""" import asyncio import functools import logging import urllib from typing import Optional from async_lru import alru_cache from defence360agent.api.server import API, APIError from defence360agent.utils.validate import IP, IPVersion logger = logging.getLogger(__name__) TIMEOUT_FOR_IPECHO_REQUEST = 5 # in seconds class IPEchoAPI(API): """Make requests to the API for obtain own IP address""" URL = "/api/ip" @classmethod @alru_cache(maxsize=3) async def get_ip(cls, ip_version: IPVersion = None) -> Optional[str]: """Return cached result for resolved IP from echo ip API""" return await cls.ip_for_version(ip_version) @classmethod @functools.lru_cache(maxsize=1) def server_ip(cls): """Return cached result for resolved IP from echo ip API""" try: return cls._get_ip() except Exception as e: raise APIError from e @classmethod async def ip_for_version( cls, ip_version: IPVersion = None ) -> Optional[str]: """Return resolved IP from echo ip API""" loop = asyncio.get_event_loop() try: ip = await asyncio.wait_for( loop.run_in_executor(None, cls._get_ip), timeout=TIMEOUT_FOR_IPECHO_REQUEST, ) if IP.type_of(ip) != ip_version: raise ValueError("Wrong ip type") return ip except (asyncio.TimeoutError, ValueError) as e: raise APIError from e @classmethod def _get_ip(cls): """Send request to API and process response""" request = urllib.request.Request(cls._BASE_URL + cls.URL) response = cls.request(request) if response.get("status") != "ok": # time inside sync executor raise APIError("Unexpected API error") return response.get("ip")