Current File : /opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/utils//ipecho.py
"""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")
